diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index 82f3652fbc7..e1d6dd1711e 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -38,6 +38,64 @@ report, so it counts concurrent writes without blocking on its own scan. ## Scheduled routine wake protocol +Historical failed ephemeral leases and terminal custom-image setup sessions can +become idle only after the host observes deletion of their exact provider +resource. Cloud checks bounded batches on already-awake instances, outside the +readiness path. The optional `environmentInspectLease` plugin RPC is read-only: +it must never reconnect, start, stop, or destroy an allocation. Older workers +without this method remain conservative. Daytona accepts its SDK's typed +not-found response; authentication and transport failures are not deletion +proof. Setup checks require the recorded provider-config fingerprint; lease +checks use the recorded plugin and scoped config, never a replacement binding. +For legacy secret references without a version manifest, only an unrotated +initial version with a matching resolved-value hash establishes credential +identity. Rotated/mutable credentials and ambient Daytona keys defer proof; +not-found under another account cannot establish historical cleanup. Inspection +also rejects an ambient API endpoint override when no endpoint was recorded. +The manager permits at most two outstanding inspection RPCs across workers. +Caller timeouts retain their slots until the actual worker reply or exit; +timeouts cannot accumulate additional provider reads in later sweeps. + +Host-owned verification metadata binds the company, environment, resource, +provider handle, optional run, and exact row version. Concurrent resume, +replacement, cleanup, or configuration changes invalidate verification. +Provider-supplied metadata cannot supply this marker. Inspection preserves +failure outcomes and history and does not create a run-continuation receipt. +Normal setup cancellation and timeout preserve the host's recorded config +fingerprint and RPC company scope for later read-only verification. +Retained allocations, unfinished runs/accounting/retries, export intents and +live setup credentials continue to block sleep. + +Pending human-only connection setup can wait asleep when no active enabled +connection of the requested purpose exists. Available access reblocks sleep so +the existing reconciliation path can run; answering a setup card continues to +use the normal admission and wake path. +The delivery worker uses the same passive-wait proof to release its local +admission hold. On an already-awake instance it keeps its existing five-second +maintenance cadence, without creating a wake alarm. This catches newly available +access even when its write does not notify the connection queue. Pending real +deliveries and their downstream wake requests still block sleep. +An owned drain pauses passive-only maintenance before it admits tracked work, +so a slow pre-sleep backup can finish with an unchanged generation. New producer +notifications still preempt the maintenance timer and invalidate that backup. + +An unconfigured Slack draft can also wait for browser setup when its matching +channel connection is disabled and still a draft. It must have no provider +identity, prior activation/activity, or setup error. Partially configured, +enabled, foreign, and other-provider endpoints remain blockers; accepted chat +deliveries retain their independent checks. +An optional Slack slash-command name is configuration only and does not create +background work. Other setup fields still retain their blockers. + +Cloud also reconciles missed routine issue-status hooks in bounded batches on +awake instances. Exact company/routine/issue/run ownership and row locks are +required. The normal routine service records completion or transient failure +from the current issue status; a resumed issue can use its existing continuation +rules. This repairs stale handoff bookkeeping without exempting unprocessed +routine runs or consuming deferred comment requests. +Lock contention and retryable transaction failures defer only that candidate; +other repairs continue, and later sweeps retry the deferred row. + Verified signed Cloud control callers may opt into `GET /api/instance/task-drain?idleSleepSafety=1&scheduledWake=1&ownerId=`. Tenant tokens, browser instance administrators, local implicit actors, and agents @@ -235,7 +293,28 @@ including token refresh after a quota response times out. Credential-recovery journals, partial writes and probes block sleep across restart. Empty company directories left after successful recovery do not; invalid paths and read failures remain unknown. Credentials and connection records are retained. -MCP, chat and other connections still block until they have a wake contract. +Disabled draft connections and saved outbound HTTP MCP connections have no +independent listener. They can permit sleep while accepted calls and credential +work still pass the other fences. MCP connections with a retained GitHub +webhook binding, local stdio processes, channels, and other active transports +remain blockers until their inbound work has a supported wake contract. + +With the issue-scheduling capability, pending questions, confirmations, +checkbox confirmations and task suggestions can be passive waits. Their +producer must have finished without retry or accounting debt. The response +audience and continuation policy stay unchanged. Unknown pending siblings, +foreign-company records and unfinished producers block sleep. A resolved card +does not make an otherwise runnable issue passive. The schedule generation +includes response identity and audience, including when only a routine owns +the next alarm. +Merge confirmations remain blockers because GitHub polling can accept them +without a human response and does not yet have a durable wake contract. + +Known board escalation actions for stranded work, missing disposition and a +lost review path can also be passive. They require a blocked source in the same +company, board ownership, no automatic wake, no monitor or timeout, and no live +execution binding. The action remains visible and unresolved. Other recovery +causes and autonomous owners remain blockers. A failed ephemeral lease is history only with successful cleanup, a release timestamp, and an exact provider termination receipt bound to its company, run, @@ -243,6 +322,10 @@ lease and resource. Its run must be finished without retry or accounting debt. Pending cleanup claims and export/stop-and-retain intents still block. Failure labels or cleanup success without that proof do not authorize sleep. Lease history is retained; this check performs no provider teardown or data updates. +A released local ephemeral bookkeeping lease needs no remote termination +receipt when its exact run has finished and both its provider and environment +are local. It must have no provider resource reference or pending cleanup, +export or retention intent. Unknown drivers and remote resources still block. Finished login-session history permits sleep only after its promotion claim is cleared and any retained provider reference has a matching released, expired, @@ -265,10 +348,19 @@ Pending deliveries, missing completion evidence, and unfinished handoffs still block. Failed deliveries without a pending target are history; other work checks still apply. -Every enabled plugin blocks sleep. A version label does not prove that arbitrary -worker code has no background activity. No plugin approval or configuration -bypass is supported. Scheduled and externally triggered work needs a durable -host wake mechanism before this restriction can be relaxed. +Processed connector-event receipts with a processing timestamp are idempotency +history. Delivered connection-intent outboxes also permit sleep once their +exact accepted wakes finish. A system-retired connection card parked behind a +pending response keeps its undelivered outbox intact; the normal response path +resumes delivery. Pending event application, unfinished wakes and all other +execution checks still block. None of these exceptions fabricates completion +or removes a durable receipt. + +Enabled plugins require the current worker's verified idle drain. A version +label does not prove that arbitrary worker code has no background activity. +No plugin approval or configuration bypass is supported. Scheduled and +externally triggered work needs a durable host wake mechanism before it can +permit sleep. This is a single-process admission protocol. A restart clears the owner. The host must serialize sleep with deploy/wake operations and bind its final check diff --git a/packages/plugins/paperclip-plugin-fake-sandbox/src/plugin.ts b/packages/plugins/paperclip-plugin-fake-sandbox/src/plugin.ts index 30e7001d5a2..4b7d7ef2738 100644 --- a/packages/plugins/paperclip-plugin-fake-sandbox/src/plugin.ts +++ b/packages/plugins/paperclip-plugin-fake-sandbox/src/plugin.ts @@ -13,6 +13,8 @@ import type { PluginEnvironmentDeleteTemplateParams, PluginEnvironmentDeleteTemplateResult, PluginEnvironmentDestroyLeaseParams, + PluginEnvironmentInspectLeaseParams, + PluginEnvironmentLeaseObservation, PluginEnvironmentExecuteParams, PluginEnvironmentExecuteResult, PluginEnvironmentGetInteractiveSetupParams, @@ -344,6 +346,10 @@ const plugin = definePlugin({ } }, + async onEnvironmentInspectLease(params: PluginEnvironmentInspectLeaseParams): Promise { + return { providerLeaseId: params.providerLeaseId, state: leases.has(params.providerLeaseId) ? "active" : "destroyed" }; + }, + async onEnvironmentDestroyLease( params: PluginEnvironmentDestroyLeaseParams, ): Promise { diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 3bd65960443..1803bcd1dbd 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -40,6 +40,46 @@ import { environmentSyncErrorData, readEnvironmentSyncErrorDiagnostic, withEnvir import manifest from "./manifest.js"; import { parseTarVerboseListingLine, splitLinkEntryOnce } from "./file-sync.js"; +describe("Daytona read-only resource inspection", () => { + const params = { driverKey: "daytona", companyId: "fixture-company", environmentId: "fixture-environment", + providerLeaseId: "sandbox-123", config: { apiKey: "fixture-not-a-key" } }; + afterEach(() => { vi.clearAllMocks(); vi.unstubAllEnvs(); __resetDaytonaSandboxHandleCacheForTest(); }); + it.each(["started", "stopped"])("observes a %s sandbox without lifecycle calls", async state => { + const sandbox = createMockSandbox({ state }); mockGet.mockResolvedValue(sandbox); + expect(await plugin.definition.onEnvironmentInspectLease!(params)).toEqual({ providerLeaseId: "sandbox-123", state: state === "stopped" ? "stopped" : "active" }); + expect(mockGet).toHaveBeenLastCalledWith("sandbox-123"); + for (const method of [sandbox.start, sandbox.stop, sandbox.delete, sandbox.archive, sandbox.refreshData]) expect(method).not.toHaveBeenCalled(); + expect(mockCreate).not.toHaveBeenCalled(); + }); + it("accepts only typed provider not-found as deletion proof", async () => { + __resetDaytonaSandboxHandleCacheForTest(); + mockGet.mockRejectedValue(new MockDaytonaNotFoundError("not found")); + expect(await plugin.definition.onEnvironmentInspectLease!(params)).toEqual({ providerLeaseId: "sandbox-123", state: "destroyed" }); + expect(await plugin.definition.onIdleDrain!(new AbortController().signal)).toBe("none"); + }); + it("does not use an ambient account credential to certify historical deletion", async () => { + expect(await plugin.definition.onEnvironmentInspectLease!({ ...params, config: {} })).toEqual({ providerLeaseId: "sandbox-123", state: "unknown" }); + expect(mockGet).not.toHaveBeenCalled(); + }); + it.each(["DAYTONA_API_URL", "DAYTONA_SERVER_URL"])("does not substitute ambient %s for the recorded API endpoint", async variable => { + vi.stubEnv(variable, "https://replacement.example.test/api"); + expect((await plugin.definition.onEnvironmentInspectLease!(params)).state).toBe("unknown"); + expect(mockGet).not.toHaveBeenCalled(); + mockGet.mockRejectedValue(new MockDaytonaNotFoundError("gone")); + expect((await plugin.definition.onEnvironmentInspectLease!({ ...params, config: { ...params.config, apiUrl: "https://recorded.example.test/api" } })).state).toBe("destroyed"); + }); + it.each(["401 unauthorized", "404 proxy not found", "network unavailable"])("does not turn %s into proof", async message => { + mockGet.mockRejectedValue(new Error(message)); + await expect(plugin.definition.onEnvironmentInspectLease!(params)).rejects.toThrow(message); + }); + it("rejects another resource and re-reads provider state on every observation", async () => { + mockGet.mockResolvedValueOnce(createMockSandbox({ id: "another-sandbox" })).mockRejectedValueOnce(new MockDaytonaNotFoundError("gone")); + expect((await plugin.definition.onEnvironmentInspectLease!(params)).state).toBe("unknown"); + expect((await plugin.definition.onEnvironmentInspectLease!(params)).state).toBe("destroyed"); + expect(mockGet).toHaveBeenCalledTimes(2); + }); +}); + describe("Daytona idle eligibility", () => { it("permits a fresh unused provider but remains conservative after provider access", async () => { __resetDaytonaSandboxHandleCacheForTest(); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 5be933bebb1..96b9c3f0d04 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -24,6 +24,8 @@ import type { PluginEnvironmentDeleteTemplateParams, PluginEnvironmentDeleteTemplateResult, PluginEnvironmentDestroyLeaseParams, + PluginEnvironmentInspectLeaseParams, + PluginEnvironmentLeaseObservation, PluginEnvironmentExecuteParams, PluginEnvironmentExecuteResult, PluginEnvironmentRunnerIngressEndpointParams, @@ -319,8 +321,8 @@ function resolveApiKey(config: DaytonaDriverConfig): string { return envApiKey; } -function createDaytonaClient(config: DaytonaDriverConfig): Daytona { - providerUsed = true; +function createDaytonaClient(config: DaytonaDriverConfig, readOnly = false): Daytona { + if (!readOnly) providerUsed = true; const clientConfig: DaytonaConfig = { apiKey: resolveApiKey(config), }; @@ -2886,6 +2888,30 @@ const plugin = definePlugin({ } }, + async onEnvironmentInspectLease( + params: PluginEnvironmentInspectLeaseParams, + ): Promise { + // An environment-variable fallback may have changed accounts since this + // historical allocation. Inspection requires explicitly scoped credentials. + if (typeof params.config.apiKey !== "string" || !params.config.apiKey.trim()) { + return { providerLeaseId: params.providerLeaseId, state: "unknown" }; + } + const config = parseDriverConfig(params.config); + if (!config.apiUrl && (process.env.DAYTONA_API_URL?.trim() || process.env.DAYTONA_SERVER_URL?.trim())) { + return { providerLeaseId: params.providerLeaseId, state: "unknown" }; + } + // Read directly from the provider. The setup-status path and cached lease + // reconnect path can start a stopped sandbox, so neither is an observation. + try { + const sandbox = await createDaytonaClient(config, true).get(params.providerLeaseId); + if (sandbox.id !== params.providerLeaseId) return { providerLeaseId: params.providerLeaseId, state: "unknown" }; + return { providerLeaseId: sandbox.id, state: sandbox.state === "stopped" ? "stopped" : "active" }; + } catch (error) { + if (error instanceof DaytonaNotFoundError) return { providerLeaseId: params.providerLeaseId, state: "destroyed" }; + throw error; + } + }, + async onEnvironmentDestroyLease( params: PluginEnvironmentDestroyLeaseParams, ): Promise { diff --git a/packages/plugins/sdk/src/define-plugin.ts b/packages/plugins/sdk/src/define-plugin.ts index f206f8c8399..3d7d81a31b5 100644 --- a/packages/plugins/sdk/src/define-plugin.ts +++ b/packages/plugins/sdk/src/define-plugin.ts @@ -55,6 +55,8 @@ import type { PluginContext, ResourceLifecycleEvent, ResourceLifecycleResult } f import type { PluginEnvironmentAcquireLeaseParams, PluginEnvironmentDestroyLeaseParams, + PluginEnvironmentInspectLeaseParams, + PluginEnvironmentLeaseObservation, PluginEnvironmentExecuteParams, PluginEnvironmentExecuteResult, PluginEnvironmentRunnerIngressEndpointParams, @@ -431,6 +433,11 @@ export interface PluginDefinition { params: PluginEnvironmentDestroyLeaseParams, ): Promise; + /** Read provider state only. This must not start or reconnect a sandbox. */ + onEnvironmentInspectLease?( + params: PluginEnvironmentInspectLeaseParams, + ): Promise; + /** Called to materialize the run workspace inside the provider lease. */ onEnvironmentRealizeWorkspace?( params: PluginEnvironmentRealizeWorkspaceParams, diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 5290d59568c..8698d6ba4bf 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -185,6 +185,8 @@ export type { PluginEnvironmentReleaseLeaseParams, PluginEnvironmentTerminationReceipt, PluginEnvironmentDestroyLeaseParams, + PluginEnvironmentInspectLeaseParams, + PluginEnvironmentLeaseObservation, PluginEnvironmentRealizeWorkspaceParams, PluginEnvironmentRealizeWorkspaceResult, PluginEnvironmentExecuteParams, diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 6e0d1f15e89..c6db765ce6a 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -686,6 +686,18 @@ export interface PluginEnvironmentTerminationReceipt { export interface PluginEnvironmentDestroyLeaseParams extends PluginEnvironmentReleaseLeaseParams {} +/** Observe this exact allocation without starting, stopping or deleting it. + * Missing resources must be distinguished from permission/transport failures. */ +export interface PluginEnvironmentInspectLeaseParams extends PluginEnvironmentDriverBaseParams { + providerLeaseId: string; + leaseMetadata?: Record; +} + +export interface PluginEnvironmentLeaseObservation { + providerLeaseId: string; + state: "active" | "stopped" | "destroyed" | "unknown"; +} + export interface PluginEnvironmentRealizeWorkspaceParams extends PluginEnvironmentDriverBaseParams { lease: PluginEnvironmentLease; workspace: { @@ -1400,6 +1412,10 @@ export interface HostToWorkerMethods { params: PluginEnvironmentDestroyLeaseParams, result: PluginEnvironmentTerminationReceipt | void, ]; + environmentInspectLease: [ + params: PluginEnvironmentInspectLeaseParams, + result: PluginEnvironmentLeaseObservation, + ]; environmentRealizeWorkspace: [ params: PluginEnvironmentRealizeWorkspaceParams, result: PluginEnvironmentRealizeWorkspaceResult, @@ -1506,6 +1522,7 @@ export const HOST_TO_WORKER_OPTIONAL_METHODS: readonly HostToWorkerMethodName[] "environmentReleaseLease", "environmentStopLease", "environmentDestroyLease", + "environmentInspectLease", "environmentRealizeWorkspace", "environmentExecute", "environmentRunnerIngressEndpoint", diff --git a/packages/plugins/sdk/src/testing.ts b/packages/plugins/sdk/src/testing.ts index 479bd998d85..cbfdca2114d 100644 --- a/packages/plugins/sdk/src/testing.ts +++ b/packages/plugins/sdk/src/testing.ts @@ -52,6 +52,8 @@ import type { PluginEnvironmentProbeParams, PluginEnvironmentProbeResult, PluginEnvironmentLease, + PluginEnvironmentInspectLeaseParams, + PluginEnvironmentLeaseObservation, PluginEnvironmentAcquireLeaseParams, PluginEnvironmentResumeLeaseParams, PluginEnvironmentReleaseLeaseParams, @@ -163,6 +165,7 @@ export interface EnvironmentEventRecord { | "resumeLease" | "releaseLease" | "stopLease" + | "inspectLease" | "destroyLease" | "realizeWorkspace" | "execute" @@ -190,6 +193,7 @@ export interface EnvironmentTestHarnessOptions extends TestHarnessOptions { onResumeLease?: (params: PluginEnvironmentResumeLeaseParams) => Promise; onReleaseLease?: (params: PluginEnvironmentReleaseLeaseParams) => Promise; onStopLease?: (params: PluginEnvironmentReleaseLeaseParams) => Promise; + onInspectLease?: (params: PluginEnvironmentInspectLeaseParams) => Promise; onDestroyLease?: (params: PluginEnvironmentDestroyLeaseParams) => Promise; onRealizeWorkspace?: (params: PluginEnvironmentRealizeWorkspaceParams) => Promise; onExecute?: (params: PluginEnvironmentExecuteParams) => Promise; @@ -217,6 +221,8 @@ export interface EnvironmentTestHarness extends TestHarness { releaseLease(params: PluginEnvironmentReleaseLeaseParams): Promise; /** Stop and preserve an allocation independently of its release policy. */ stopLease(params: PluginEnvironmentReleaseLeaseParams): Promise; + /** Observe an allocation without starting or modifying it. */ + inspectLease(params: PluginEnvironmentInspectLeaseParams): Promise; /** Invoke the environment driver's destroyLease hook. */ destroyLease(params: PluginEnvironmentDestroyLeaseParams): Promise; /** Invoke the environment driver's realizeWorkspace hook. */ @@ -2775,6 +2781,9 @@ export function createEnvironmentTestHarness(options: EnvironmentTestHarnessOpti async stopLease(params) { return callHook("stopLease", driver.onStopLease, params, "onStopLease"); }, + async inspectLease(params) { + return callHook("inspectLease", driver.onInspectLease, params, "onInspectLease"); + }, async destroyLease(params) { return callHook("destroyLease", driver.onDestroyLease, params, "onDestroyLease"); }, diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 4812054450a..3de92ac5216 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -93,6 +93,7 @@ import type { RefreshExternalObjectsParams, PluginEnvironmentAcquireLeaseParams, PluginEnvironmentDestroyLeaseParams, + PluginEnvironmentInspectLeaseParams, PluginEnvironmentExecuteParams, PluginEnvironmentRunnerIngressEndpointParams, PluginEnvironmentSyncInParams, @@ -1699,6 +1700,10 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost case "environmentDestroyLease": return handleEnvironmentDestroyLease(params as PluginEnvironmentDestroyLeaseParams); + case "environmentInspectLease": + if (!plugin.definition.onEnvironmentInspectLease) throw methodNotImplemented("environmentInspectLease"); + return plugin.definition.onEnvironmentInspectLease(params as PluginEnvironmentInspectLeaseParams); + case "environmentRealizeWorkspace": return handleEnvironmentRealizeWorkspace(params as PluginEnvironmentRealizeWorkspaceParams); @@ -1802,6 +1807,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (plugin.definition.onEnvironmentReleaseLease) supportedMethods.push("environmentReleaseLease"); if (plugin.definition.onEnvironmentStopLease) supportedMethods.push("environmentStopLease"); if (plugin.definition.onEnvironmentDestroyLease) supportedMethods.push("environmentDestroyLease"); + if (plugin.definition.onEnvironmentInspectLease) supportedMethods.push("environmentInspectLease"); if (plugin.definition.onEnvironmentRealizeWorkspace) supportedMethods.push("environmentRealizeWorkspace"); if (plugin.definition.onEnvironmentExecute) supportedMethods.push("environmentExecute"); if (plugin.definition.onEnvironmentRunnerIngressEndpoint) { diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index 0c759222c3c..61f523d84ad 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -418,6 +418,24 @@ describe("worker configChanged cross-tenant guard", () => { } finally { worker.stop(); } }); + it.each([false, true])("advertises read-only inspection only when implemented: %s", async supported => { + let inspections = 0, mutations = 0; + const worker = makeWorker(definePlugin({ async setup() {}, + async onEnvironmentReleaseLease() { mutations++; }, + async onEnvironmentDestroyLease() { mutations++; }, + ...(supported ? { async onEnvironmentInspectLease() { inspections++; return { providerLeaseId: "allocation", state: "destroyed" as const }; } } : {}), + })); + try { + const initialized = await worker.initialize() as { supportedMethods: string[] }; + expect(initialized.supportedMethods.includes("environmentInspectLease")).toBe(supported); + const result = worker.callWorker("environmentInspectLease", { driverKey: "fixture", companyId: "company", environmentId: "environment", providerLeaseId: "allocation", config: {} }); + if (supported) await expect(result).resolves.toEqual({ providerLeaseId: "allocation", state: "destroyed" }); + else await expect(result).rejects.toThrow(); + expect(inspections).toBe(supported ? 1 : 0); + expect(mutations).toBe(0); + } finally { worker.stop(); } + }); + it("fails closed when a second, distinct company's config would overwrite a single-tenant worker", async () => { const applied: Array<{ companyId: string | null; token: unknown }> = []; const plugin = definePlugin({ diff --git a/server/src/__tests__/delivery-work-coordinator.test.ts b/server/src/__tests__/delivery-work-coordinator.test.ts index d3f7590a79b..9531d21deba 100644 --- a/server/src/__tests__/delivery-work-coordinator.test.ts +++ b/server/src/__tests__/delivery-work-coordinator.test.ts @@ -2,10 +2,12 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { notifyDeliveryWork, DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; -import { idleWorkSnapshot } from "../services/task-admission.js"; +import { beginIdleTrackedWork, idleWorkSnapshot, isIdleTaskDrainActive, readTaskDrain, startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; +import type { Db } from "@paperclipai/db"; const coordinators: ReturnType[] = []; -afterEach(async () => { for (const c of coordinators.splice(0)) await c.stop(); vi.useRealTimers(); }); +afterEach(async () => { stopTaskDrain(); for (const c of coordinators.splice(0)) await c.stop(); vi.useRealTimers(); }); function setup() { vi.useFakeTimers(); let outcome = "committed"; @@ -89,6 +91,43 @@ describe("delivery work coordinator", () => { expect(t.run).toHaveBeenCalledTimes(3); expect(idleWorkSnapshot().active).toBe(0); }); + it.each([false, true])("pauses passive maintenance through a slow owned backup while preserving new work (%s)", async newWork => { + const s = setup(), t = task(); + s.canRun.mockImplementation(() => !isIdleTaskDrainActive()); + s.canReconcile.mockReturnValue(true); + await s.coordinator.register(DELIVERY_QUEUES.connection, { + ...t, nextRunAt: () => Date.now() + 5000, pausePassiveMaintenance: true, + }).ready; + const drain = startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + const getStatus = () => { + const current = readTaskDrain(new Date()); + return { ...current, draining: current !== null, startedAt: current?.startedAt ?? null, + expiresAt: current?.expiresAt ?? null, activeRuns: 0, pendingWakes: 0 }; + }; + const db = { transaction: async (fn: (tx: unknown) => Promise) => + fn({ execute: async () => [{ blocked: false }] }) } as unknown as Db; + let entered!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const checkpoint = async () => { + const finish = beginIdleTrackedWork(); entered(); + try { await new Promise(resolve => setTimeout(resolve, 6000)); return true; } + finally { finish(); } + }; + expect(idleWorkSnapshot().active).toBe(0); + const scan = readIdleSleepSafety(db, getStatus, Date.now, drain.ownerId, async () => "none", undefined, checkpoint); + await Promise.race([started, scan.then(result => { throw new Error(`Backup did not start: ${JSON.stringify(result)}`); })]); + expect(idleWorkSnapshot().active).toBe(1); + if (newWork) await s.enqueue(DELIVERY_QUEUES.connection); + await vi.advanceTimersByTimeAsync(6000); + expect(await scan).toEqual({ version: 1, backgroundWork: newWork ? "unknown" : "none" }); + expect(t.run).toHaveBeenCalledTimes(1); + expect(t.hasPending).toHaveBeenCalledTimes(newWork ? 2 : 1); + expect(idleWorkSnapshot().active).toBe(0); + stopTaskDrain(); + await vi.advanceTimersByTimeAsync(5000); + expect(t.run).toHaveBeenCalledTimes(2); + expect(s.onError).not.toHaveBeenCalled(); + }); it("fences unresolved writes until the server settles, then scans and releases the idle hold", async () => { const s = setup(), t = task(); await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready; diff --git a/server/src/__tests__/environment-custom-images-service.test.ts b/server/src/__tests__/environment-custom-images-service.test.ts index fd9ba739570..6422cec2f91 100644 --- a/server/src/__tests__/environment-custom-images-service.test.ts +++ b/server/src/__tests__/environment-custom-images-service.test.ts @@ -77,6 +77,7 @@ const ALL_CUSTOM_IMAGE_WORKER_METHODS = [ function createWorkerManager(options?: { workerMethods?: readonly string[] }) { const supportedMethods = options?.workerMethods ?? ALL_CUSTOM_IMAGE_WORKER_METHODS; const call = vi.fn(async (_pluginId: string, method: string, params: Record) => { + if (method === "environmentInspectLease") return { providerLeaseId: params.providerLeaseId, state: "destroyed" }; if (method === "environmentStartInteractiveSetup") { return { providerLeaseId: `lease-${params.sessionId}`, @@ -199,6 +200,28 @@ describeEmbeddedPostgres("environmentCustomImageService", () => { return { companyId, environmentId }; } + it.each(["supported", "old_worker", "changed_config"])("inspects historical setup read-only with exact config and capability gates: %s", async mode => { + const fixture = await seed(); + const worker = createWorkerManager({ workerMethods: [...ALL_CUSTOM_IMAGE_WORKER_METHODS, + ...(mode === "old_worker" ? [] : ["environmentInspectLease"])] }); + const service = environmentCustomImageService(db, { pluginWorkerManager: worker }); + const created = await service.startSetupSession({ environmentId: fixture.environmentId, + actor: { userId: "fixture-user" }, now: new Date("2099-01-01T00:00:00Z") }); + await service.cancelSetupSession({ sessionId: created.session.id }); + const [session] = await db.select().from(environmentCustomImageSetupSessions).where(eq(environmentCustomImageSetupSessions.id, created.session.id)); + expect(session!.metadata).toMatchObject({ setupRpcCompanyId: created.session.metadata!.setupRpcCompanyId, + sourceEnvironmentConfigFingerprint: created.session.metadata!.sourceEnvironmentConfigFingerprint }); + if (mode === "changed_config") await db.update(environments).set({ config: { provider: "fake-plugin", image: "replacement:base" } }).where(eq(environments.id, fixture.environmentId)); + worker.call.mockClear(); + const observed = await service.inspectSetupResource(session!); + if (mode === "supported") { + expect(observed).toEqual({ providerLeaseId: session!.providerLeaseId, state: "destroyed" }); + expect(worker.call).toHaveBeenCalledExactlyOnceWith(expect.any(String), "environmentInspectLease", expect.objectContaining({ + companyId: session!.metadata!.setupRpcCompanyId, providerLeaseId: session!.providerLeaseId, + }), 15_000); + } else { expect(observed).toBeUndefined(); expect(worker.call).not.toHaveBeenCalled(); } + }); + it("starts, refreshes, finishes, refreshes again, and rolls back setup sessions", async () => { const { environmentId } = await seed(); const workerManager = createWorkerManager(); @@ -385,6 +408,8 @@ describeEmbeddedPostgres("environmentCustomImageService", () => { reason: "user_cancelled", }); expect(cancelled.status).toBe("cancelled"); + expect(cancelled.metadata).toMatchObject({ setupRpcCompanyId: started.session.metadata!.setupRpcCompanyId, + sourceEnvironmentConfigFingerprint: started.session.metadata!.sourceEnvironmentConfigFingerprint }); expect(await service.getActiveTemplate({ environmentId })).toBeNull(); const expired = await service.startSetupSession({ @@ -399,6 +424,8 @@ describeEmbeddedPostgres("environmentCustomImageService", () => { expect(cleanup).toMatchObject({ scanned: 1, timedOut: 1, failed: 0 }); const timedOut = await service.getSessionById(expired.session.id); expect(timedOut?.status).toBe("timed_out"); + expect(timedOut?.metadata).toMatchObject({ setupRpcCompanyId: expired.session.metadata!.setupRpcCompanyId, + sourceEnvironmentConfigFingerprint: expired.session.metadata!.sourceEnvironmentConfigFingerprint }); }); it("rejects interactive setup when the provider declares it but the worker omits a setup method", async () => { diff --git a/server/src/__tests__/fixtures/plugin-worker-delayed.cjs b/server/src/__tests__/fixtures/plugin-worker-delayed.cjs index 48a68cfcc51..baae987cb7c 100644 --- a/server/src/__tests__/fixtures/plugin-worker-delayed.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-delayed.cjs @@ -8,6 +8,9 @@ const rl = readline.createInterface({ input: process.stdin, crlfDelay: Infinity, }); +const heldInspections = []; +let inspectionCount = 0; +let activeInspections = 0; rl.on("line", (line) => { if (!line.trim()) return; @@ -20,7 +23,7 @@ rl.on("line", (line) => { id: message.id, result: { ok: true, - supportedMethods: ["environmentExecute", "configChanged"], + supportedMethods: ["environmentExecute", "configChanged", "environmentInspectLease", "health"], }, }); return; @@ -45,6 +48,9 @@ rl.on("line", (line) => { } if (method === "configChanged") { + if (message.params.config.releaseInspections) { + for (const complete of heldInspections.splice(0)) complete(); + } setTimeout(() => send({ jsonrpc: "2.0", id: message.id, ...(message.params.config.fail @@ -54,6 +60,20 @@ rl.on("line", (line) => { return; } + if (method === "environmentInspectLease") { + inspectionCount++; + activeInspections++; + const complete = () => { activeInspections--; send({ jsonrpc: "2.0", id: message.id, + result: { providerLeaseId: message.params.providerLeaseId, state: "destroyed" } }); }; + if (message.params.leaseMetadata?.holdInspection) heldInspections.push(complete); + else complete(); + return; + } + if (method === "health") { + send({ jsonrpc: "2.0", id: message.id, result: { status: "ok", details: { inspectionCount, activeInspections } } }); + return; + } + if (method === "shutdown") { send({ jsonrpc: "2.0", diff --git a/server/src/__tests__/historical-cleanup-verification.test.ts b/server/src/__tests__/historical-cleanup-verification.test.ts new file mode 100644 index 00000000000..37bc2dd6692 --- /dev/null +++ b/server/src/__tests__/historical-cleanup-verification.test.ts @@ -0,0 +1,126 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, companies, createDb, environmentCustomImageSetupSessions, environmentLeases, environments, heartbeatRuns } from "@paperclipai/db"; +import { historicalCleanupVerificationService } from "../services/historical-cleanup-verification.js"; +import { withoutCleanupVerification } from "../services/cleanup-verification-proof.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const now = Date.UTC(2020, 0, 1), ownerId = randomUUID(); +const owned = (): IdleSleepDrainStatus => ({ ownerId, draining: true, startedAt: new Date(now - 1000), + expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 }); +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping historical cleanup database tests: ${support.reason}`); +(support.supported ? describe : describe.skip)("historical cleanup verification", () => { + let db: ReturnType, database: Awaited>; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("historical-cleanup-"); db = createDb(database.connectionString); }, 60_000); + afterEach(async () => { await db.execute(sql`SET client_min_messages = warning; TRUNCATE companies, plugins, environments CASCADE`); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + const read = () => readIdleSleepSafety(db, owned, () => now, ownerId, async () => "none", undefined, undefined, true, true); + const state = async () => (await read()).backgroundWork; + async function seed(adHoc = false) { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Cleanup fixture", issuePrefix: `T${companyId.slice(0, 6).toUpperCase()}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Idle fixture", role: "engineer", status: "idle", adapterType: "process" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "failed", finishedAt: new Date(now) }); + const [environment] = await db.insert(environments).values({ name: "Remote fixture", driver: "sandbox" }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId, environmentId: environment!.id, + heartbeatRunId: adHoc ? null : runId, status: "failed", leasePolicy: "ephemeral", provider: "daytona", + providerLeaseId: randomUUID(), releasedAt: new Date(now), cleanupStatus: "success", metadata: { driver: "sandbox" } }).returning(); + return { companyId, runId, environment: environment!, lease: lease! }; + } + function verifier(leaseState = "destroyed", setupState = "destroyed") { + return historicalCleanupVerificationService(db, { + inspectLease: async row => ({ providerLeaseId: row.providerLeaseId, state: leaseState }), + inspectSetup: async row => ({ providerLeaseId: row.providerLeaseId, state: setupState }), + }); + } + it.each([false, true])("accepts exact deletion proof without changing outcomes or granting run authority (ad-hoc %s)", async adHoc => { + const f = await seed(adHoc); expect(await state()).toBe("present"); + const service = verifier(); expect(await service.sweep()).toMatchObject({ checked: 1, verified: 1, failed: 0 }); + expect(await state()).toBe("none"); + const [current] = await db.select().from(environmentLeases); + expect(current).toMatchObject({ status: "failed", cleanupStatus: "success", heartbeatRunId: adHoc ? null : f.runId }); + expect(current!.metadata).not.toHaveProperty("remoteExecutionTermination"); + expect(await service.sweep()).toMatchObject({ checked: 0, verified: 0 }); + }); + it.each(["active", "stopped", "unknown"])("retains %s allocations", async status => { + await seed(); expect(await verifier(status).sweep()).toMatchObject({ verified: 0, deferred: 1 }); + expect(await state()).toBe("present"); + }); + it.each(["old_worker", "wrong_id", "provider_error"])("does not manufacture proof from %s", async mode => { + await seed(); + const service = historicalCleanupVerificationService(db, { inspectSetup: async () => undefined, + inspectLease: async () => { if (mode === "provider_error") throw new Error("private credential failure"); + return mode === "old_worker" ? undefined : { providerLeaseId: "wrong", state: "destroyed" }; } }); + expect((await service.sweep()).verified).toBe(0); expect(await state()).toBe("present"); + }); + it.each(["resume", "resource", "environment", "metadata", "run", "accounting", "retry"])("loses verification when %s changes during provider inspection", async change => { + const f = await seed(); + const service = historicalCleanupVerificationService(db, { inspectSetup: async () => undefined, inspectLease: async row => { + if (change === "resume") await db.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, row.id)); + if (change === "resource") await db.update(environmentLeases).set({ providerLeaseId: "replacement" }).where(eq(environmentLeases.id, row.id)); + if (change === "environment") await db.update(environmentLeases).set({ environmentId: null }).where(eq(environmentLeases.id, row.id)); + if (change === "metadata") await db.update(environmentLeases).set({ metadata: { driver: "sandbox", nativeWorkspaceExportResume: {} } }).where(eq(environmentLeases.id, row.id)); + if (["run", "accounting", "retry"].includes(change)) await db.update(heartbeatRuns).set(change === "run" ? { status: "running", finishedAt: null } + : change === "accounting" ? { costAccountingPending: true } : { scheduledRetryAt: new Date(now + 1000) }).where(eq(heartbeatRuns.id, f.runId)); + return { providerLeaseId: row.providerLeaseId, state: "destroyed" }; + } }); + expect((await service.sweep()).verified).toBe(0); expect(await state()).toBe("present"); + }); + it.each(["retained", "pending_cleanup"])("does not inspect a resource with another %s owner", async status => { + const f = await seed(); + await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId: f.environment.id, + provider: "daytona", providerLeaseId: f.lease.providerLeaseId, status }); + expect(await verifier().sweep()).toMatchObject({ checked: 0, verified: 0 }); + expect(await state()).toBe("present"); + }); + it("invalidates a previously accepted proof when the lease row changes", async () => { + const f = await seed(); await verifier().sweep(); expect(await state()).toBe("none"); + await db.update(environmentLeases).set({ updatedAt: new Date(0) }).where(eq(environmentLeases.id, f.lease.id)); + expect(await state()).toBe("present"); + }); + it("keeps provider work tracked through the observation and coalesces concurrent sweeps", async () => { + await seed(); let finish!: () => void; let entered!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const paused = new Promise(resolve => { finish = resolve; }); + const inspectLease = vi.fn(async (row: typeof environmentLeases.$inferSelect) => { entered(); await paused; return { providerLeaseId: row.providerLeaseId, state: "destroyed" }; }); + const service = historicalCleanupVerificationService(db, { inspectLease, inspectSetup: async () => undefined }); + const before = idleWorkSnapshot().active; const first = service.sweep(); const second = service.sweep(); + expect(first).toBe(second); await started; expect(idleWorkSnapshot().active).toBe(before + 1); + expect(await state()).toBe("present"); finish(); await first; + expect(inspectLease).toHaveBeenCalledTimes(1); expect(idleWorkSnapshot().active).toBe(before); + }); + async function setup() { + const f = await seed(); await db.delete(environmentLeases); + const [row] = await db.insert(environmentCustomImageSetupSessions).values({ environmentId: f.environment.id, + provider: "daytona", providerLeaseId: randomUUID(), status: "promoted", finishedAt: new Date(now), metadata: { setupRpcCompanyId: f.companyId } }).returning(); + return { ...f, setup: row! }; + } + it("requires deletion proof for terminal image sessions and preserves their history", async () => { + const f = await setup(); expect(await state()).toBe("present"); + expect(await verifier().sweep()).toMatchObject({ verified: 1 }); expect(await state()).toBe("none"); + expect((await db.select().from(environmentCustomImageSetupSessions))[0]).toMatchObject({ status: "promoted", providerLeaseId: f.setup.providerLeaseId }); + }); + it.each(["secret", "live", "provider", "scope"])("rejects image-session %s changes during inspection", async change => { + await setup(); + const service = historicalCleanupVerificationService(db, { inspectLease: async () => undefined, inspectSetup: async row => { + await db.update(environmentCustomImageSetupSessions).set(change === "secret" ? { connectionSecretRef: "secret://fixture" } + : change === "live" ? { status: "waiting_for_user", finishedAt: null } + : change === "provider" ? { provider: "other" } : { metadata: { setupRpcCompanyId: randomUUID() } }).where(eq(environmentCustomImageSetupSessions.id, row.id)); + return { providerLeaseId: row.providerLeaseId, state: "destroyed" }; + } }); + expect((await service.sweep()).verified).toBe(0); expect(await state()).toBe("present"); + }); + it("keeps image setup blocked when a retained lease owns the same allocation", async () => { + const f = await setup(); await db.insert(environmentLeases).values({ companyId: f.companyId, + environmentId: f.environment.id, provider: "daytona", providerLeaseId: f.setup.providerLeaseId, status: "retained" }); + expect(await verifier().sweep()).toMatchObject({ checked: 0 }); expect(await state()).toBe("present"); + }); +}); + +it("removes host-owned cleanup evidence from provider metadata", () => { + expect(withoutCleanupVerification({ paperclipCleanupVerification: { state: "destroyed" }, driver: "sandbox" })).toEqual({ driver: "sandbox" }); +}); diff --git a/server/src/__tests__/historical-provider-credentials.test.ts b/server/src/__tests__/historical-provider-credentials.test.ts new file mode 100644 index 00000000000..256610cb3df --- /dev/null +++ b/server/src/__tests__/historical-provider-credentials.test.ts @@ -0,0 +1,61 @@ +import { createHash, randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { companies, companySecrets, companySecretVersions, createDb } from "@paperclipai/db"; +import type { SandboxEnvironmentConfig } from "@paperclipai/shared"; +const resolve = vi.hoisted(() => vi.fn()); +vi.mock("../services/secrets.js", () => ({ secretService: () => ({ resolveSecretValueForSandboxCleanup: resolve }) })); +import { historicalProviderCredentials } from "../services/historical-provider-credentials.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const value = "fixture-original-provider-key"; +const hash = createHash("sha256").update(value).digest("hex"); +const schema = { properties: { apiKey: { type: "string", format: "secret-ref" } } }; +const context = { issueId: null, heartbeatRunId: null }; +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping provider credential database tests: ${support.reason}`); +(support.supported ? describe : describe.skip)("historical provider credential identity", () => { + let db: ReturnType, database: Awaited>; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("historical-credentials-"); db = createDb(database.connectionString); }, 60_000); + afterEach(async () => { resolve.mockReset(); await db.execute(sql`SET client_min_messages = warning; TRUNCATE companies CASCADE`); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + async function seed() { + const companyId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Credential fixture", issuePrefix: `T${companyId.slice(0, 6)}` }); + const [secret] = await db.insert(companySecrets).values({ companyId, key: "provider-key", name: "Provider key" }).returning(); + await db.insert(companySecretVersions).values({ secretId: secret!.id, version: 1, material: {}, valueSha256: hash, fingerprintSha256: hash }); + resolve.mockResolvedValue(value); return { companyId, secret: secret! }; + } + const inspect = (companyId: string, apiKey: unknown, manifest = schema) => historicalProviderCredentials(db, companyId, + { provider: "daytona", apiKey, reuseLease: false } as SandboxEnvironmentConfig, manifest, context); + it("uses exact recorded literal credentials without resolving another account", async () => { + const f = await seed(); expect(await inspect(f.companyId, value)).toMatchObject({ apiKey: value }); expect(resolve).not.toHaveBeenCalled(); + }); + it("defers missing credentials instead of relying on an ambient account", async () => { + const f = await seed(); expect(await inspect(f.companyId, undefined)).toBeUndefined(); expect(resolve).not.toHaveBeenCalled(); + }); + it.each([false, true])("pins and checks an unrotated initial secret version (binding %s)", async binding => { + const f = await seed(); expect(await inspect(f.companyId, binding ? { type: "secret_ref", secretId: f.secret.id, version: "latest" } : f.secret.id)).toMatchObject({ apiKey: value }); + expect(resolve).toHaveBeenCalledExactlyOnceWith(f.companyId, f.secret.id, 1, { ...context, configPath: "apiKey" }); + }); + it.each(["rotated", "rolled_back", "foreign", "revoked", "deleted", "unknown_manifest", "different_pin"])("defers uncertain credential identity: %s", async mode => { + const f = await seed(); let companyId = f.companyId; + if (["rotated", "rolled_back"].includes(mode)) { + await db.insert(companySecretVersions).values({ secretId: f.secret.id, version: 2, material: {}, valueSha256: hash, fingerprintSha256: hash }); + if (mode === "rotated") await db.update(companySecrets).set({ latestVersion: 2 }).where(eq(companySecrets.id, f.secret.id)); + } + if (mode === "foreign") companyId = randomUUID(); + if (mode === "revoked") await db.update(companySecretVersions).set({ revokedAt: new Date() }); + if (mode === "deleted") await db.update(companySecrets).set({ status: "deleted" }); + const key = mode === "different_pin" ? { type: "secret_ref", secretId: f.secret.id, version: 2 } : f.secret.id; + expect(await inspect(companyId, key, mode === "unknown_manifest" ? {} as typeof schema : schema)).toBeUndefined(); + expect(resolve).not.toHaveBeenCalled(); + }); + it("rejects externally mutable secret material whose current bytes changed accounts", async () => { + const f = await seed(); resolve.mockResolvedValue("fixture-replacement-account-key"); + expect(await inspect(f.companyId, f.secret.id)).toBeUndefined(); + }); + it("propagates credential read errors instead of calling them resource deletion", async () => { + const f = await seed(); resolve.mockRejectedValue(new Error("credential backend unavailable")); + await expect(inspect(f.companyId, f.secret.id)).rejects.toThrow("credential backend unavailable"); + }); +}); diff --git a/server/src/__tests__/idle-connection-history.test.ts b/server/src/__tests__/idle-connection-history.test.ts new file mode 100644 index 00000000000..43b3508543e --- /dev/null +++ b/server/src/__tests__/idle-connection-history.test.ts @@ -0,0 +1,156 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, agentWakeupRequests, chatEndpoints, chatSlackRegistrations, companies, connectionEventDeliveries, connectionGrants, connectionIntentDeliveries, + createDb, environmentLeases, environments, heartbeatRuns, issues, issueThreadInteractions, toolApplications, toolConnections, +} from "@paperclipai/db"; +import { beginIdleTrackedWork } from "../services/task-admission.js"; +import { connectionIntentDeliveryService } from "../services/connection-intent-delivery.js"; +import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const now = Date.UTC(2020, 0, 1); const ownerId = randomUUID(); +const owned = (): IdleSleepDrainStatus => ({ ownerId, draining: true, startedAt: new Date(now - 1000), + expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 }); +const none = { version: 1, backgroundWork: "none" }; const present = { version: 1, backgroundWork: "present" }; +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping connection-history database tests: ${support.reason}`); +(support.supported ? describe : describe.skip)("idle connection and local lease history", () => { + let db: ReturnType; let database: Awaited>; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("idle-connection-history-"); db = createDb(database.connectionString); }, 60_000); + afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + const read = () => readIdleSleepSafety(db, owned, () => now, ownerId, async () => "none", undefined, undefined, true, true); + async function seed(status = "done") { + const companyId = randomUUID(); const agentId = randomUUID(); const runId = randomUUID(); const issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Connection history fixture", issuePrefix: `T${companyId.slice(0, 6).toUpperCase()}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Idle owner", role: "engineer", status: "idle", adapterType: "process" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Fixture", status, assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "succeeded", finishedAt: new Date(now) }); + return { companyId, agentId, runId, issueId }; + } + async function delivery(f: Awaited>, deliveredAt: Date | null = new Date(now)) { + const [card] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + kind: "connection_intent", status: "expired", resolvedAt: new Date(now), sourceRunId: f.runId, + payload: { requestingAgentId: f.agentId }, result: { version: 1, outcome: "expired", connectionId: randomUUID() } }).returning(); + const [outbox] = await db.insert(connectionIntentDeliveries).values({ companyId: f.companyId, interactionId: card!.id, deliveredAt }).returning(); + return { card: card!, outbox: outbox! }; + } + it("retains processed event history while allowing sleep", async () => { + const f = await seed(); + const [event] = await db.insert(connectionEventDeliveries).values({ companyId: f.companyId, provider: "github", providerDeliveryId: randomUUID(), event: "push", status: "processed", processedAt: new Date(now) }).returning(); + expect(await read()).toEqual(none); expect(await db.select().from(connectionEventDeliveries)).toEqual([event]); + }); + it.each(["received", "failed", "missing_receipt"])("retains %s event work", async state => { + const f = await seed(); + await db.insert(connectionEventDeliveries).values({ companyId: f.companyId, provider: "github", providerDeliveryId: randomUUID(), event: "push", status: state === "missing_receipt" ? "processed" : state as "received" | "failed", processedAt: state === "missing_receipt" ? null : new Date(now) }); + expect(await read()).toEqual(present); + }); + it("keeps downstream work awake after event application finishes", async () => { + const f = await seed(); await db.insert(connectionEventDeliveries).values({ companyId: f.companyId, provider: "github", providerDeliveryId: randomUUID(), event: "push", status: "processed", processedAt: new Date(now) }); + await db.update(heartbeatRuns).set({ status: "queued", finishedAt: null }).where(eq(heartbeatRuns.id, f.runId)); + expect(await read()).toEqual(present); + }); + it("permits a delivered no-op without deleting its outbox receipt", async () => { + const d = await delivery(await seed()); expect(await read()).toEqual(none); + expect(await db.select().from(connectionIntentDeliveries)).toEqual([d.outbox]); + }); + it.each(["pending", "unfinished_wake", "queued_wake", "coalesced_unfinished"])("retains %s connection delivery work", async state => { + const f = await seed(); const d = await delivery(f, state === "pending" ? null : new Date(now)); + if (state !== "pending") await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + idempotencyKey: `connection-intent:${d.card.id}:expired`, status: state === "queued_wake" ? "queued" : state === "coalesced_unfinished" ? "coalesced" : "completed", + finishedAt: state === "coalesced_unfinished" ? new Date(now) : null }); + expect(await read()).toEqual(present); + }); + it.each(["completed", "coalesced"])("permits a delivered connection after its exact %s wake finishes", async status => { + const f = await seed(); const d = await delivery(f); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", status, runId: f.runId, + idempotencyKey: `connection-intent:${d.card.id}:expired`, finishedAt: new Date(now) }); + expect(await read()).toEqual(none); + }); + it("parks an undelivered system retirement behind its still-pending response without marking delivery complete", async () => { + const f = await seed("in_review"); const d = await delivery(f, null); + const [question] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, kind: "ask_user_questions", status: "pending", + effectiveResolverPolicy: "human_only", sourceRunId: f.runId, payload: {} }).returning(); + expect(await read()).toEqual(none); + const worker = connectionIntentDeliveryService(db, { wakeup: async () => null }); + expect(await worker.hasPending()).toBe(false); + expect(worker.nextRunAt()).toBeTypeOf("number"); + await worker.sweepPending(); + expect(await db.select().from(connectionIntentDeliveries)).toEqual([d.outbox]); + await db.update(issueThreadInteractions).set({ status: "answered", resolvedAt: new Date(now) }).where(eq(issueThreadInteractions.id, question!.id)); + expect(await worker.hasPending()).toBe(true); + expect(worker.nextRunAt()).toBeNull(); + expect(await read()).toEqual(present); + }); + async function connection(f: Awaited>, change: Partial = {}) { + const [app] = await db.insert(toolApplications).values({ companyId: f.companyId, name: "Outbound tool fixture", type: "mcp_http" }).returning(); + const [c] = await db.insert(toolConnections).values({ companyId: f.companyId, applicationId: app!.id, uid: randomUUID(), name: "Saved remote tools", + connectionPurpose: "tool", transport: "mcp_remote", status: "active", enabled: true, ...change }).returning(); return c!; + } + it("allows saved outbound MCP configuration while preserving active-call admission", async () => { + const c = await connection(await seed()); expect(await read()).toEqual(none); + const finish = beginIdleTrackedWork(); try { expect(await read()).toEqual({ version: 1, backgroundWork: "unknown" }); } finally { finish(); } + expect(await db.select().from(toolConnections)).toEqual([c]); + }); + it("retains inbound GitHub bindings until their own wake contract is supported", async () => { + const f = await seed(); const c = await connection(f); + await db.insert(connectionGrants).values({ companyId: f.companyId, connectionId: c.id, kind: "organization", providerTenant: { github: { + userId: "fixture", login: "fixture", installationCount: 1, repositoryCount: 1, repositorySelection: "all", installationIds: ["fixture"], installationOwnerLogins: ["fixture"], + } } }); expect(await read()).toEqual(present); + }); + it.each(["tool", "channel", "ai"] as const)("permits an inert disabled %s draft", async purpose => { + await connection(await seed(), { connectionPurpose: purpose, transport: purpose === "channel" ? "chat_sdk" : purpose === "ai" ? "runtime_auth" : "mcp_remote", status: "draft", enabled: false }); + expect(await read()).toEqual(none); + }); + it.each(["local_stdio", "rest_api"] as const)("retains unsupported active %s integrations", async transport => { + await connection(await seed(), { transport }); expect(await read()).toEqual(present); + }); + async function slackDraft(command?: string) { + const f = await seed(); const c = await connection(f, { connectionPurpose: "channel", transport: "chat_sdk", status: "draft", enabled: false }); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId: f.companyId, connectionId: c.id, + assignedAgentId: f.agentId, publicId: randomUUID(), provider: "slack", + setup: { step: "provider_setup", ...(command ? { command } : {}) } }).returning(); + return { ...f, connection: c, endpoint: endpoint! }; + } + it.each([undefined, "/paperclip-fixture"])("permits an unconfigured disabled Slack draft without losing setup state (%s)", async command => { + const f = await slackDraft(command); expect(await read()).toEqual(none); + expect(await db.select().from(chatEndpoints)).toEqual([f.endpoint]); + }); + it.each(["active", "provider_identity", "previous_activity", "connected_step", "health", "unsupported_provider", "enabled_connection", "setup_in_progress", "registration", "unknown_command_shape"])("retains draft integration with %s", async mode => { + const f = await slackDraft(); + if (mode === "active") await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (mode === "provider_identity") await db.update(chatEndpoints).set({ botExternalId: "fixture-bot" }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (mode === "previous_activity") await db.update(chatEndpoints).set({ lastEventAt: new Date(now) }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (mode === "connected_step") await db.update(chatEndpoints).set({ setup: { step: "test" } }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (mode === "health") await db.update(chatEndpoints).set({ healthMessage: "Provider setup pending" }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (mode === "unsupported_provider") await db.update(chatEndpoints).set({ provider: "discord" }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (mode === "enabled_connection") await db.update(toolConnections).set({ enabled: true }).where(eq(toolConnections.id, f.connection.id)); + if (mode === "setup_in_progress") await db.update(chatEndpoints).set({ setup: { step: "provider_setup", slackSetupMethod: "managed" } }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (mode === "registration") await db.insert(chatSlackRegistrations).values({ endpointId: f.endpoint.id, companyId: f.companyId, + requestId: randomUUID(), status: "creating", manifest: {}, manifestHash: "fixture-hash", callbackUri: "https://example.test/callback", createdByUserId: "fixture-user" }); + if (mode === "unknown_command_shape") await db.update(chatEndpoints).set({ setup: sql`'{"step":"provider_setup","command":42}'::jsonb` }).where(eq(chatEndpoints.id, f.endpoint.id)); + expect(await read()).toEqual(present); + }); + async function localLease() { + const f = await seed(); const [environment] = await db.insert(environments).values({ name: "Local fixture", driver: "local" }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId: environment!.id, heartbeatRunId: f.runId, + status: "failed", leasePolicy: "ephemeral", provider: "local", providerLeaseId: null, releasedAt: new Date(now), metadata: { driver: "local" } }).returning(); + return { ...f, lease: lease!, environment: environment! }; + } + it("permits released local bookkeeping after the owning run finishes", async () => { + const f = await localLease(); expect(await read()).toEqual(none); + expect(await db.select().from(environmentLeases)).toEqual([f.lease]); + }); + it.each(["provider_resource", "missing_release", "live_run", "unknown_driver", "pending_export", "retained"])("retains local lease work with %s evidence", async kind => { + const f = await localLease(); + if (kind === "provider_resource") await db.update(environmentLeases).set({ providerLeaseId: "unexpected-resource" }).where(eq(environmentLeases.id, f.lease.id)); + if (kind === "missing_release") await db.update(environmentLeases).set({ releasedAt: null }).where(eq(environmentLeases.id, f.lease.id)); + if (kind === "live_run") await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "unknown_driver") await db.update(environments).set({ driver: "ssh" }).where(eq(environments.id, f.environment.id)); + if (kind === "pending_export") await db.update(environmentLeases).set({ metadata: { driver: "local", nativeWorkspaceExportResume: {} } }).where(eq(environmentLeases.id, f.lease.id)); + if (kind === "retained") await db.update(environmentLeases).set({ leasePolicy: "retain_on_failure" }).where(eq(environmentLeases.id, f.lease.id)); + expect(await read()).toEqual(present); + }); +}); diff --git a/server/src/__tests__/idle-response-waits.test.ts b/server/src/__tests__/idle-response-waits.test.ts new file mode 100644 index 00000000000..58dafafad9a --- /dev/null +++ b/server/src/__tests__/idle-response-waits.test.ts @@ -0,0 +1,170 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, companies, createDb, heartbeatRuns, issues, issueRecoveryActions, + issueThreadInteractions, routines, routineTriggers, + toolApplications, toolConnections, +} from "@paperclipai/db"; +import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; +import { getMergeConfirmationPullRequestReferences } from "../services/issue-thread-interactions.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const now = Date.UTC(2020, 0, 1); +const ownerId = randomUUID(); +const owned = (): IdleSleepDrainStatus => ({ ownerId, draining: true, startedAt: new Date(now - 1000), + expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 }); +const none = { version: 1, backgroundWork: "none" }; +const present = { version: 1, backgroundWork: "present" }; +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping response-wait database tests: ${support.reason}`); +(support.supported ? describe : describe.skip)("idle response-driven waits", () => { + let db: ReturnType; + let database: Awaited>; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("idle-response-waits-"); + db = createDb(database.connectionString); + }, 60_000); + afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + const read = (schedules = true) => readIdleSleepSafety(db, owned, () => now, ownerId, + async () => "none", undefined, undefined, schedules, schedules); + + async function seed(status = "in_review") { + const companyId = randomUUID(); const agentId = randomUUID(); const runId = randomUUID(); const issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Response fixture", issuePrefix: `T${companyId.slice(0, 6).toUpperCase()}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Idle owner", role: "engineer", status: "idle", adapterType: "process" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Awaiting a response", status, assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", + status: "succeeded", finishedAt: new Date(now), contextSnapshot: { issueId } }); + return { companyId, agentId, runId, issueId }; + } + async function wait(f: Awaited>, change: Partial = {}) { + const [row] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + kind: "ask_user_questions", status: "pending", sourceRunId: f.runId, effectiveResolverPolicy: "human_only", + continuationPolicy: "wake_assignee", payload: { version: 1 }, ...change }).returning(); + return row!; + } + + it.each(["anyone", "not_creator", "human_only"] as const)("preserves the %s response audience while permitting an idle wait", async policy => { + const f = await seed(); const card = await wait(f, { effectiveResolverPolicy: policy }); + expect(await read(false)).toEqual(present); + expect(await read()).toEqual(none); + expect(await db.select().from(issueThreadInteractions)).toEqual([card]); + expect((await db.select().from(issues))[0]?.status).toBe("in_review"); + }); + it.each(["ask_user_questions", "request_confirmation", "request_checkbox_confirmation", "suggest_tasks"])( + "permits a persisted %s wait after its producer finishes", async kind => { + await wait(await seed("in_progress"), { kind }); + expect(await read()).toEqual(none); + }); + it.each(["human_only", "anyone"])("keeps automatically polled merge confirmations awake with %s audience", async policy => { + const f = await seed("in_progress"); + const card = { kind: "request_confirmation", title: "Merge PaperclipAI/paperclip#39?", summary: null, + payload: { version: 1, prompt: "Merge the pull request?" } }; + expect(getMergeConfirmationPullRequestReferences(card)).toHaveLength(1); + await wait(f, { ...card, effectiveResolverPolicy: policy }); + expect(await read()).toEqual(present); + }); + it("does not treat an assigned task or an answered interaction as a passive wait", async () => { + const f = await seed(); expect(await read()).toEqual(present); + const card = await wait(f); expect(await read()).toEqual(none); + await db.update(issueThreadInteractions).set({ status: "answered", resolvedAt: new Date(now) }).where(eq(issueThreadInteractions.id, card.id)); + expect(await read()).toEqual(present); + }); + it("allows human-only connection setup without reusable access, and reblocks when access appears", async () => { + const f = await seed(); + await wait(f, { kind: "connection_intent", addresseeUserId: "fixture-user", payload: { + version: 1, serviceSlug: "fixture-service", requestingAgentId: f.agentId, + } }); + expect(await read()).toEqual(none); + const [app] = await db.insert(toolApplications).values({ companyId: f.companyId, + type: "mcp_http", name: "Fixture service" }).returning(); + await db.insert(toolConnections).values({ companyId: f.companyId, applicationId: app!.id, + uid: randomUUID(), name: "Reusable access", connectionPurpose: "tool", transport: "mcp_remote", status: "active", enabled: true }); + expect(await read()).toEqual(present); + }); + it.each(["wrong_assignee", "unknown_purpose", "no_addressee", "agent_resolver"])("retains unsafe connection wait: %s", async mode => { + const f = await seed(); + await wait(f, { kind: "connection_intent", addresseeUserId: mode === "no_addressee" ? null : "fixture-user", + effectiveResolverPolicy: mode === "agent_resolver" ? "anyone" : "human_only", payload: { + version: 1, serviceSlug: "fixture-service", requestingAgentId: mode === "wrong_assignee" ? randomUUID() : f.agentId, + ...(mode === "unknown_purpose" ? { purpose: "future" } : {}), + } }); + expect(await read()).toEqual(present); + }); + it.each([ + { kind: "connection_intent" }, { kind: "unknown" }, { continuationPolicy: "automatic" }, + { effectiveResolverPolicy: "unknown" }, { resolvedAt: new Date(now) }, + ])("keeps unsupported waiting work awake: %j", async change => { + const f = await seed(); await wait(f); await wait(f, change); + expect(await read()).toEqual(present); + }); + it("does not borrow another company's interaction or ignore a foreign pending sibling", async () => { + const f = await seed(); const other = await seed("done"); + await wait(f, { companyId: other.companyId, sourceRunId: other.runId }); + expect(await read()).toEqual(present); + await wait(f); expect(await read()).toEqual(present); + }); + it.each([ + { status: "running", finishedAt: null }, { finishedAt: null }, { costAccountingPending: true }, + { scheduledRetryAt: new Date(now + 1000) }, + ])("keeps a response producer's unfinished work awake: %j", async change => { + const f = await seed(); await wait(f); + await db.update(heartbeatRuns).set(change).where(eq(heartbeatRuns.id, f.runId)); + expect(await read()).toEqual(present); + }); + it.each([ + { checkoutRunId: randomUUID() }, { executionRunId: randomUUID() }, + { executionState: { status: "pending" } }, { monitorWakeRequestedAt: new Date(now) }, + ])("does not let a response card hide issue execution: %j", async change => { + const f = await seed(); await wait(f); + // Checkout and execution fields reference real run rows. + const patch = { ...change }; + if ("checkoutRunId" in patch) patch.checkoutRunId = f.runId; + if ("executionRunId" in patch) patch.executionRunId = f.runId; + await db.update(issues).set(patch).where(eq(issues.id, f.issueId)); + expect(await read()).toEqual(present); + }); + it("includes response identity and audience in the future schedule generation", async () => { + const f = await seed(); const card = await wait(f); + const [routine] = await db.insert(routines).values({ companyId: f.companyId, title: "Later work", status: "active", assigneeAgentId: f.agentId }).returning(); + await db.insert(routineTriggers).values({ companyId: f.companyId, routineId: routine!.id, kind: "schedule", + cronExpression: "*/15 * * * *", timezone: "UTC", nextRunAt: new Date(now + 600_000), enabled: true, setupPending: false }); + const before = await read(); expect(before.version).toBe(2); + await db.update(issueThreadInteractions).set({ effectiveResolverPolicy: "anyone" }).where(eq(issueThreadInteractions.id, card.id)); + const after = await read(); expect(after.scheduledWake?.nextWakeAt).toBe(before.scheduledWake?.nextWakeAt); + expect(after.scheduledWake?.generation).not.toBe(before.scheduledWake?.generation); + }); + + async function boardHold(f: Awaited>, change: Partial = {}) { + const [row] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "stranded_assigned_issue", cause: "stranded_assigned_issue", status: "active", ownerType: "board", + fingerprint: randomUUID(), nextAction: "Board decision required", wakePolicy: { type: "board_escalation", reason: "stranded_assigned_issue", preservesSourceAssignee: true }, + ...change }).returning(); + return row!; + } + it.each(["active", "escalated"])("retains visible %s board escalation history without keeping compute running", async status => { + const f = await seed("blocked"); const action = await boardHold(f, { status }); + expect(await read()).toEqual(none); + expect(await db.select().from(issueRecoveryActions)).toEqual([action]); + }); + it.each([ + { ownerType: "agent" }, { ownerType: "system" }, { cause: "native_workspace_finalization_owner_unverified" }, + { timeoutAt: new Date(now + 60_000) }, { monitorPolicy: {} }, + { wakePolicy: { type: "wake_owner" } }, { wakePolicy: { type: "board_escalation", retryAt: "2099-01-01" } }, + { wakePolicy: null }, { resolvedAt: new Date(now) }, + ])("keeps automatic or unproved recovery work awake: %j", async change => { + await boardHold(await seed("blocked"), change); + expect(await read()).toEqual(present); + }); + it("rejects a foreign recovery source and an escalation whose source still runs", async () => { + const f = await seed("blocked"); const other = await seed("done"); + const action = await boardHold(f, { companyId: other.companyId }); + expect(await read()).toEqual(present); + await db.update(issueRecoveryActions).set({ companyId: f.companyId }).where(eq(issueRecoveryActions.id, action.id)); + expect(await read()).toEqual(none); + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.runId)); + expect(await read()).toEqual(present); + }); +}); diff --git a/server/src/__tests__/idle-sleep-safety.test.ts b/server/src/__tests__/idle-sleep-safety.test.ts index ad10c180c79..befce2fac67 100644 --- a/server/src/__tests__/idle-sleep-safety.test.ts +++ b/server/src/__tests__/idle-sleep-safety.test.ts @@ -592,13 +592,13 @@ if (!support.supported) console.warn(`Skipping idle sleep Postgres tests: ${supp }, ); - it("retains remote MCP connections alongside inert AI credentials", async () => { + it("allows saved outbound MCP connections alongside inert AI credentials", async () => { const { companyId, connection } = await seedAiConnection(); await db.insert(toolConnections).values({ companyId, applicationId: connection.applicationId, name: "Remote tools", uid: randomUUID(), connectionPurpose: "tool", transport: "mcp_remote", status: "active", enabled: true, }); - expect(await read()).toEqual(present); + expect(await read()).toEqual(none); }); async function seedFinishedFailedLease() { diff --git a/server/src/__tests__/plugin-inspection-concurrency.test.ts b/server/src/__tests__/plugin-inspection-concurrency.test.ts new file mode 100644 index 00000000000..41c57323dbb --- /dev/null +++ b/server/src/__tests__/plugin-inspection-concurrency.test.ts @@ -0,0 +1,38 @@ +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, it } from "vitest"; +import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; +import { createPluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +it("bounds actual provider reads across caller timeouts, plugins, retries, and worker exit", async () => { + const baseline = idleWorkSnapshot().active; + const manager = createPluginWorkerManager(); + const manifest: PaperclipPluginManifestV1 = { id: "fixture.inspect", apiVersion: 1, version: "1.0.0", + displayName: "Inspection fixture", categories: ["automation"], capabilities: [], entrypoints: { worker: "worker.cjs" } }; + const params = { driverKey: "fixture", companyId: "fixture-company", environmentId: "fixture-environment", + config: {}, providerLeaseId: "fixture-resource", leaseMetadata: { holdInspection: true } }; + try { + for (const pluginId of ["fixture.inspect.a", "fixture.inspect.b"]) await manager.startWorker(pluginId, { + entrypointPath: path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/plugin-worker-delayed.cjs"), + manifest, config: {}, instanceInfo: { instanceId: "fixture-instance", hostVersion: "1.0.0" }, apiVersion: 1, hostHandlers: {}, + }); + await Promise.all(["fixture.inspect.a", "fixture.inspect.b"].map(pluginId => + expect(manager.call(pluginId, "environmentInspectLease", params, 20)).rejects.toThrow("timed out"))); + expect(idleWorkSnapshot().active).toBe(baseline + 2); + for (let retry = 0; retry < 4; retry++) await expect(manager.call("fixture.inspect.a", "environmentInspectLease", params, 20)).rejects.toThrow("concurrency limit"); + for (const pluginId of ["fixture.inspect.a", "fixture.inspect.b"]) expect(await manager.call(pluginId, "health", {})).toMatchObject({ + details: { inspectionCount: 1, activeInspections: 1 }, + }); + await manager.call("fixture.inspect.a", "configChanged", { companyId: "fixture-company", config: { releaseInspections: true } }); + expect(idleWorkSnapshot().active).toBe(baseline + 1); + await expect(manager.call("fixture.inspect.a", "environmentInspectLease", params, 20)).rejects.toThrow("timed out"); + await expect(manager.call("fixture.inspect.b", "environmentInspectLease", params, 20)).rejects.toThrow("concurrency limit"); + await manager.stopWorker("fixture.inspect.b"); + expect(idleWorkSnapshot().active).toBe(baseline + 1); + await expect(manager.call("fixture.inspect.a", "environmentInspectLease", { ...params, leaseMetadata: {} }, 1000)).resolves.toEqual({ + providerLeaseId: "fixture-resource", state: "destroyed", + }); + } finally { await manager.stopAll(); } + expect(idleWorkSnapshot().active).toBe(baseline); +}, 15_000); diff --git a/server/src/__tests__/routine-status-reconciliation.test.ts b/server/src/__tests__/routine-status-reconciliation.test.ts new file mode 100644 index 00000000000..eba0718ce25 --- /dev/null +++ b/server/src/__tests__/routine-status-reconciliation.test.ts @@ -0,0 +1,87 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { companies, createDb, issues, routineRuns, routines } from "@paperclipai/db"; +import { routineService } from "../services/routines.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping routine status database tests: ${support.reason}`); +(support.supported ? describe : describe.skip)("routine status reconciliation", () => { + let db: ReturnType, database: Awaited>; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("routine-status-"); db = createDb(database.connectionString); }, 60_000); + afterEach(async () => { vi.restoreAllMocks(); await db.execute(sql`SET client_min_messages = warning; TRUNCATE companies CASCADE`); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + async function seed(status = "blocked") { + const companyId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Routine fixture", issuePrefix: `T${companyId.slice(0, 6)}` }); + const [routine] = await db.insert(routines).values({ companyId, title: "Routine fixture" }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Execution fixture", status, + originKind: "routine_execution", originId: routine!.id, originRunId: runId }).returning(); + const [run] = await db.insert(routineRuns).values({ id: runId, companyId, routineId: routine!.id, + linkedIssueId: issue!.id, source: "schedule", status: "issue_created", triggerPayload: { retainedInput: true } }).returning(); + return { companyId, routine: routine!, issue: issue!, run: run! }; + } + it.each(["done", "blocked", "cancelled"])("repairs missed %s status using normal owning-service transitions", async status => { + const f = await seed(status); const service = routineService(db); + expect(await service.reconcileExecutionRunStatuses()).toEqual({ scanned: 1, settled: 1, deferred: 0 }); + const [run] = await db.select().from(routineRuns); + expect(run!.status).toBe(status === "done" ? "completed" : "failed"); + expect(run!.completedAt).toBeInstanceOf(Date); expect(run!.triggerPayload).toMatchObject({ retainedInput: true }); + expect((await db.select().from(issues))[0]).toEqual(f.issue); + expect(await service.reconcileExecutionRunStatuses()).toEqual({ scanned: 0, settled: 0, deferred: 0 }); + if (status !== "done") { + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id)); + await service.syncRunStatusForIssue(f.issue.id); + expect((await db.select().from(routineRuns))[0]).toMatchObject({ status: "issue_created", completedAt: null }); + } + }); + it.each(["active_issue", "wrong_origin", "wrong_routine", "wrong_run", "wrong_company", "wrong_routine_company", "different_status"])("does not settle an unsupported handoff: %s", async mode => { + const f = await seed(); + if (mode === "active_issue") await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id)); + if (mode === "wrong_origin") await db.update(issues).set({ originKind: "manual" }).where(eq(issues.id, f.issue.id)); + if (mode === "wrong_routine") await db.update(issues).set({ originId: randomUUID() }).where(eq(issues.id, f.issue.id)); + if (mode === "wrong_run") await db.update(issues).set({ originRunId: randomUUID() }).where(eq(issues.id, f.issue.id)); + if (mode === "wrong_company") { const other = await seed("done"); await db.update(routineRuns).set({ companyId: other.companyId }).where(eq(routineRuns.id, f.run.id)); } + if (mode === "wrong_routine_company") { const other = await seed("done"); + await db.update(routineRuns).set({ routineId: other.routine.id }).where(eq(routineRuns.id, f.run.id)); + await db.update(issues).set({ originId: other.routine.id }).where(eq(issues.id, f.issue.id)); } + if (mode === "different_status") await db.update(routineRuns).set({ status: "received" }).where(eq(routineRuns.id, f.run.id)); + const before = await db.select().from(routineRuns); + const result = await routineService(db).reconcileExecutionRunStatuses(); + expect(result.settled).toBe(["wrong_company", "wrong_routine_company"].includes(mode) ? 1 : 0); + expect((await db.select().from(routineRuns)).find(row => row.id === f.run.id)).toEqual(before.find(row => row.id === f.run.id)); + if (mode === "wrong_routine_company") await expect(routineService(db).syncRunStatusForIssue(f.issue.id)).resolves.toBeNull(); + }); + it.each(["resume", "retarget", "routine_transfer"])("rechecks %s after candidate selection and before repair", async mode => { + const f = await seed(); const transaction = db.transaction.bind(db); + vi.spyOn(db, "transaction").mockImplementationOnce(async callback => { + if (mode === "routine_transfer") { + const otherId = randomUUID(); await db.insert(companies).values({ id: otherId, name: "Other fixture", issuePrefix: `T${otherId.slice(0, 6)}` }); + await db.update(routines).set({ companyId: otherId }).where(eq(routines.id, f.routine.id)); + } else await db.update(issues).set(mode === "resume" ? { status: "in_progress" } : { originRunId: randomUUID() }).where(eq(issues.id, f.issue.id)); + return transaction(callback); + }); + expect(await routineService(db).reconcileExecutionRunStatuses()).toEqual({ scanned: 1, settled: 0, deferred: 0 }); + expect((await db.select().from(routineRuns))[0]).toEqual(f.run); + }); + it("repairs another routine while one row is locked, then retries the deferred owner", async () => { + const [held, other] = [await seed(), await seed()].sort((a, b) => a.run.id.localeCompare(b.run.id)); + let entered!: () => void, release!: () => void; + const locked = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const holding = db.transaction(async tx => { + await tx.select().from(routines).where(eq(routines.id, held!.routine.id)).for("update"); + entered(); await gate; + }); + try { + await Promise.race([locked, holding.then(() => { throw new Error("Lock transaction ended before entry"); })]); + const service = routineService(db); + expect(await service.reconcileExecutionRunStatuses()).toEqual({ scanned: 2, settled: 1, deferred: 1 }); + expect((await db.select().from(routineRuns)).find(row => row.id === other!.run.id)?.status).toBe("failed"); + expect((await db.select().from(routineRuns)).find(row => row.id === held!.run.id)?.status).toBe("issue_created"); + release(); await holding; + expect(await service.reconcileExecutionRunStatuses()).toEqual({ scanned: 1, settled: 1, deferred: 0 }); + } finally { release(); await holding; } + }); +}); diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts index ed37e0b6337..8c8f4a53309 100644 --- a/server/src/__tests__/satisfied-connection-intents.test.ts +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -14,6 +14,9 @@ import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-wor import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { legacyExecutionNeedsReconciliationWithEvidence } from "../services/legacy-execution-recovery.js"; import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; const support = await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; @@ -86,6 +89,60 @@ describePostgres("already available connection requests", () => { } finally { unsubscribe(); } }); + it("releases a created human setup card's worker hold, then resumes when access becomes available", async () => { + const f = await seed(); + await db.delete(issueThreadInteractions); + await db.update(toolConnections).set({ status: "draft", enabled: false }).where(eq(toolConnections.id, f.connection.id)); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "succeeded", finishedAt: new Date(), contextSnapshot: { issueId: f.issueId } }).returning(); + let cardId = ""; + const wakeup = vi.fn(async () => { + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + status: "queued", idempotencyKey: `connection-intent:${cardId}:expired` }); + return null; + }); + const service = connectionIntentDeliveryService(db, { wakeup }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { errors.push(error); } }); + let sawPassive!: () => void, sawDelivered!: () => void; + const passive = new Promise(resolve => { sawPassive = resolve; }); + const delivered = new Promise(resolve => { sawDelivered = resolve; }); + const wait = async (event: Promise) => { + let timer: ReturnType | undefined; + try { await Promise.race([event, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error("Connection worker did not settle")), 8000); })]); } + finally { clearTimeout(timer); } + await new Promise(resolve => setImmediate(resolve)); + }; + try { + await coordinator.register(DELIVERY_QUEUES.connection, { retryMs: 5000, run: () => service.sweepPending(), + hasPending: async () => { const pending = await service.hasPending(); + if (!pending && service.nextRunAt() !== null) sawPassive(); + if (!pending && wakeup.mock.calls.length > 0) sawDelivered(); + return pending; + }, nextRunAt: service.nextRunAt, + }).ready; + const card = await issueThreadInteractionService(db).createConnectionIntent({ id: f.issueId, companyId: f.companyId }, { + payload: f.intent.payload, sourceRunId: source!.id, addresseeUserId: "connection-owner", idempotencyKey: "passive-setup-card", + }); + cardId = card.id; + await wait(passive); + expect(errors).toEqual([]); expect(idleWorkSnapshot().active).toBe(0); + expect((await db.select().from(issueThreadInteractions))[0].status).toBe("pending"); + const now = Date.now(), ownerId = randomUUID(); + const drain = (): IdleSleepDrainStatus => ({ ownerId, draining: true, startedAt: new Date(now - 1000), + expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 }); + expect(await readIdleSleepSafety(db, drain, Date.now, ownerId, async () => "none", undefined, undefined, true, true)).toMatchObject({ backgroundWork: "none" }); + // This mutation has no queue notification: awake-only maintenance must + // still discover new access; after sleep startup performs the same scan. + await db.update(toolConnections).set({ status: "active", enabled: true }).where(eq(toolConnections.id, f.connection.id)); + await wait(delivered); + expect(errors).toEqual([]); expect(wakeup).toHaveBeenCalledOnce(); + expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).not.toBeNull(); + expect((await db.select().from(agentWakeupRequests))[0]).toMatchObject({ status: "queued", idempotencyKey: `connection-intent:${cardId}:expired` }); + expect(await readIdleSleepSafety(db, drain, Date.now, ownerId, async () => "none", undefined, undefined, true, true)).toMatchObject({ backgroundWork: "present" }); + } finally { await coordinator.stop(); } + }, 20_000); + it.each(["wrong_identity", "revoked", "disabled", "removed_member", "denied_policy"] as const)("rechecks %s after the card retires", async (reason) => { const f = await seed(); const run = await retry(f); await satisfiedConnectionIntentService(db).sweepPending(); diff --git a/server/src/index.ts b/server/src/index.ts index 264adcd5589..409b35da93f 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -107,6 +107,8 @@ import { deliverNativeQuestionResponse } from "./services/native-runtime/native- import { queueIssueAssignmentWakeup } from "./services/issue-assignment-wakeup.js"; import { createSecretProposalsService } from "./services/secret-proposals.js"; import { environmentRuntimeService } from "./services/environment-runtime.js"; +import { historicalCleanupVerificationService } from "./services/historical-cleanup-verification.js"; +import { isCloudManagedInstance } from "./services/cloud-instance.js"; import { createDbAdapterAuthSessionStore } from "./services/device-login-service.js"; import { createDeviceLoginReaper, @@ -1327,6 +1329,7 @@ async function startServerWithDatabaseTeardown( retryMs: config.heartbeatSchedulerIntervalMs, run: () => service.sweepPending(), hasPending: () => service.hasPending(), + ...(queue === DELIVERY_QUEUES.connection ? { nextRunAt: connectionDeliveries.nextRunAt, pausePassiveMaintenance: true } : {}), })); // Preserve the existing task-scoped activity fast path. Claims in the // delivery service arbitrate with the coordinator's startup/recovery sweep. @@ -1337,10 +1340,27 @@ async function startServerWithDatabaseTeardown( .catch(err => logger.error({ err }, "post-commit chat completion delivery failed"))); }); server.on("close", unsubscribeChatCompletions); + const cleanupInspectionRuntime = environmentRuntimeService(db as any, { pluginWorkerManager }); + const environmentCustomImages = environmentCustomImageService(db as any, { pluginWorkerManager }); + const historicalCleanup = historicalCleanupVerificationService(db as any, { + inspectLease: lease => cleanupInspectionRuntime.inspectLeaseResource(lease.id), + inspectSetup: setup => environmentCustomImages.inspectSetupResource(setup), + }); + let nextHistoricalCleanupCheck = 0; + const scheduleHistoricalCleanupVerification = () => { + if (!isCloudManagedInstance() || isIdleTaskDrainActive() || isWarmStandby() || heartbeatSchedulerStopped || + Date.now() < nextHistoricalCleanupCheck) return; + nextHistoricalCleanupCheck = Date.now() + 5 * 60_000; + // Provider reads never delay readiness or the ordinary cleanup sweep. + trackHeartbeatSchedulerWork(historicalCleanup.sweep().then(result => { + if (result.verified > 0) logger.info(result, "verified historical provider cleanup"); + }).catch(() => logger.warn("historical provider cleanup verification deferred"))); + }; const runEnvironmentLeaseCleanupSweep = (backoffMs: number) => environmentLeaseCleanupHeartbeat .sweepPendingCleanupLeases({ backoffMs }) .then((result) => { + scheduleHistoricalCleanupVerification(); if (result.destroyed > 0 || result.capped > 0) { logger.info(result, "environment lease cleanup sweep retried orphan sandbox teardowns"); } @@ -1405,8 +1425,16 @@ async function startServerWithDatabaseTeardown( drainHeartbeatExecutionFinalizers = () => heartbeat.drainActiveRunExecutions(); prepareHotRestartShutdown = heartbeat.prepareHotRestartShutdown; - const environmentCustomImages = environmentCustomImageService(db as any, { pluginWorkerManager }); const routines = routineService(db as any, { pluginWorkerManager }); + let nextRoutineStatusReconcile = 0; + const scheduleRoutineStatusReconcile = () => { + if (!isCloudManagedInstance() || isWarmStandby() || isIdleTaskDrainActive() || heartbeatSchedulerStopped || + Date.now() < nextRoutineStatusReconcile) return; + nextRoutineStatusReconcile = Date.now() + 5 * 60_000; + trackHeartbeatSchedulerWork(routines.reconcileExecutionRunStatuses().then(result => { + if (result.settled > 0) logger.info(result, "reconciled routine execution issue status"); + }).catch(() => logger.warn("routine execution status reconciliation deferred"))); + }; const statusCards = statusCardService(db as any); const issues = issueService(db as any); const mergedPullRequestConfirmations = issueThreadInteractionService(db as any, { @@ -1725,6 +1753,7 @@ async function startServerWithDatabaseTeardown( // captures the authoritative set of running heartbeat rows. trackHeartbeatSchedulerWork((async () => { if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; + scheduleRoutineStatusReconcile(); trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired().catch((err: unknown) => { logger.error({ err }, "decision expiry sweep failed"); })); diff --git a/server/src/services/cleanup-verification-proof.ts b/server/src/services/cleanup-verification-proof.ts new file mode 100644 index 00000000000..13bb9930742 --- /dev/null +++ b/server/src/services/cleanup-verification-proof.ts @@ -0,0 +1,32 @@ +export const CLEANUP_VERIFICATION_KEY = "paperclipCleanupVerification"; +export const CLEANUP_VERIFICATION_SCHEMA = "paperclip.cleanup-verification.v1"; +export const CLEANUP_ROW_STAMP_FORMAT = 'YYYY-MM-DD"T"HH24:MI:SS.US'; + +/** Provider-supplied metadata cannot create host verification evidence. */ +export function withoutCleanupVerification(metadata: Record | null | undefined) { + if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) return null; + const { [CLEANUP_VERIFICATION_KEY]: _ignored, ...rest } = metadata; + return rest; +} + +export function cleanupRowStamp(alias: string): string { + return `to_char(${alias}.updated_at AT TIME ZONE 'UTC', '${CLEANUP_ROW_STAMP_FORMAT}')`; +} + +/** Fixed aliases only; never interpolate request input into these predicates. */ +export function verifiedResourceCleanup(alias: string, kind: "lease" | "image_setup"): string { + const proof = `${alias}.metadata->'${CLEANUP_VERIFICATION_KEY}'`; + const company = kind === "lease" ? `${alias}.company_id::text` + : `COALESCE(NULLIF(btrim(${alias}.metadata->>'setupRpcCompanyId'), ''), 'instance')`; + return `jsonb_typeof(${proof}) = 'object' + AND ${proof}->>'schema' = '${CLEANUP_VERIFICATION_SCHEMA}' AND ${proof}->>'kind' = '${kind}' + AND ${proof}->>'resourceId' = ${alias}.id::text AND ${proof}->>'companyId' = ${company} + AND ${proof}->>'environmentId' = ${alias}.environment_id::text + AND ${proof}->>'provider' = ${alias}.provider + AND ${proof}->>'providerLeaseId' = ${alias}.provider_lease_id + AND ${alias}.provider_lease_id IS NOT NULL AND ${alias}.provider_lease_id <> '' + AND ${proof}->>'state' = 'destroyed' + AND ${proof}->>'rowUpdatedAt' = ${cleanupRowStamp(alias)} + ${kind === "lease" ? `AND ${proof} ? 'heartbeatRunId' + AND ${proof}->'heartbeatRunId' = COALESCE(to_jsonb(${alias}.heartbeat_run_id::text), 'null'::jsonb)` : ""}`; +} diff --git a/server/src/services/connection-intent-delivery.ts b/server/src/services/connection-intent-delivery.ts index 3a5c8dfba44..561cd6e7ba0 100644 --- a/server/src/services/connection-intent-delivery.ts +++ b/server/src/services/connection-intent-delivery.ts @@ -7,6 +7,7 @@ import type { heartbeatService } from "./heartbeat.js"; import { issueService } from "./issues.js"; import { issueRecoveryActionService } from "./issue-recovery-actions.js"; import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; +import { connectionDeliveryWaitingForResponse, connectionIntentWaitingForResponse } from "./idle-work-proof.js"; type Heartbeat = ReturnType; export async function wakeConnectionIntentAfterResolution( @@ -215,18 +216,31 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick 0) return true; - return (await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions) - .innerJoin(issues, and(eq(issues.id, issueThreadInteractions.issueId), eq(issues.companyId, issueThreadInteractions.companyId))) - .where(and(eq(issueThreadInteractions.kind, "connection_intent"), eq(issueThreadInteractions.status, "pending"), - inArray(issues.status, ["in_progress", "in_review"]), isNull(issues.assigneeUserId))) - .limit(1)).length > 0; + const [state] = await db.execute<{ active: boolean; present: boolean }>(sql` + WITH pending_deliveries AS ( + SELECT (${sql.raw(connectionDeliveryWaitingForResponse('d'))}) IS TRUE AS passive + FROM connection_intent_deliveries d WHERE d.delivered_at IS NULL + ), pending_intents AS ( + SELECT (${sql.raw(connectionIntentWaitingForResponse('card', 'task'))}) IS TRUE AS passive + FROM issue_thread_interactions card LEFT JOIN issues task ON task.id = card.issue_id AND task.company_id = card.company_id + WHERE card.kind = 'connection_intent' AND card.status = 'pending' + AND (task.id IS NULL OR (task.status IN ('in_progress', 'in_review') AND task.assignee_user_id IS NULL)) + ) SELECT EXISTS (SELECT 1 FROM pending_deliveries WHERE NOT passive + UNION ALL SELECT 1 FROM pending_intents WHERE NOT passive) AS active, + EXISTS (SELECT 1 FROM pending_deliveries UNION ALL SELECT 1 FROM pending_intents) AS present`); + if (!state || typeof state.active !== "boolean" || typeof state.present !== "boolean") throw new Error("Connection work state is unknown"); + // Passive records need no admitted-work hold or wake alarm. While the + // process is already awake, keep the existing reconciliation cadence so + // new access or responses resume work even without a queue notification. + maintenanceAt = !state.active && state.present ? Date.now() + 5_000 : null; + return state.active; } - return { deliver, hasPending, tryDeliver: async (id: string) => { try { await deliver(id); } catch { /* Persisted delivery remains due after its lease. */ } }, sweepPending: async () => { + return { deliver, hasPending, nextRunAt: () => maintenanceAt, tryDeliver: async (id: string) => { try { await deliver(id); } catch { /* Persisted delivery remains due after its lease. */ } }, sweepPending: async () => { await satisfiedIntents.sweepPending(); - const rows = await db.select().from(connectionIntentDeliveries).where(and(isNull(connectionIntentDeliveries.deliveredAt), lte(connectionIntentDeliveries.nextAttemptAt, new Date()))) + const rows = await db.select().from(connectionIntentDeliveries).where(and(isNull(connectionIntentDeliveries.deliveredAt), lte(connectionIntentDeliveries.nextAttemptAt, new Date()), + sql`(${sql.raw(connectionDeliveryWaitingForResponse('connection_intent_deliveries'))}) IS NOT TRUE`)) .orderBy(asc(connectionIntentDeliveries.nextAttemptAt)).limit(50); let failed = 0; for (const row of rows) { try { await deliver(row.interactionId); } catch { failed += 1; } } diff --git a/server/src/services/delivery-work-coordinator.ts b/server/src/services/delivery-work-coordinator.ts index 2efb832250d..08f8dc47153 100644 --- a/server/src/services/delivery-work-coordinator.ts +++ b/server/src/services/delivery-work-coordinator.ts @@ -26,6 +26,9 @@ export function createDeliveryWorkCoordinator(input: { onSettled?: (hints?: ReadonlySet) => void; // Optional maintenance/retry deadline restored from durable state. nextRunAt?: () => number | null; + // An empty queue's maintenance may wait until dispatch admission reopens. + // Notifications and unresolved producer writes still reconcile in drain. + pausePassiveMaintenance?: boolean; }) { if (stopped) throw new Error("Delivery coordinator is stopped"); if (workers.has(queue)) throw new Error(`Delivery worker already registered: ${queue}`); @@ -35,6 +38,7 @@ export function createDeliveryWorkCoordinator(input: { let workerStopped = false; let controller: AbortController | null = null; let finishIntent: (() => void) | undefined; + let pendingWork = true; function schedule(delay: number, deadline?: number) { if (workerStopped) return; const at = deadline ?? Date.now() + delay; @@ -47,13 +51,22 @@ export function createDeliveryWorkCoordinator(input: { scheduler.cancel(queue); scheduledAt = null; if (!input.canRun() && !(input.canReconcile?.() ?? false)) { schedule(task.retryMs); return; } + if (!input.canRun() && task.pausePassiveMaintenance && !pendingWork && !dirty && + !finishIntent && !databaseWorkPending(input.owner, queue)) { + // A timer alone must not change the idle generation during a slow + // backup. Retry without querying or admitting work; a notification + // still preempts this deadline and retains its normal intent hold. + schedule(task.retryMs); + return; + } dirty = false; const finish = beginIdleTrackedWork(); const attempt = new AbortController(); controller = attempt; running = Promise.resolve().then(() => reconcileDatabaseWork(input.owner, queue)) .then(() => workerStopped || !input.canRun() ? undefined : task.run(attempt.signal)).then(() => workerStopped ? false : task.hasPending()).then(pending => { - if (pending || databaseWorkPending(input.owner, queue)) schedule(task.retryMs); + pendingWork = pending || databaseWorkPending(input.owner, queue); + if (pendingWork) schedule(task.retryMs); else { finishIntent?.(); finishIntent = undefined; } const next = task.nextRunAt?.(); if (next != null) { diff --git a/server/src/services/environment-custom-images.ts b/server/src/services/environment-custom-images.ts index f377b2bd1a2..7f81a935086 100644 --- a/server/src/services/environment-custom-images.ts +++ b/server/src/services/environment-custom-images.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { and, desc, eq, inArray, lte } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -22,6 +23,7 @@ import type { PluginEnvironmentCaptureTemplateResult, PluginEnvironmentInteractiveSetupConnectionPayload, PluginEnvironmentInteractiveSetupSession, + PluginEnvironmentLeaseObservation, PluginEnvironmentTemplateRefKind, } from "@paperclipai/plugin-sdk"; import { conflict, notFound, unprocessable } from "../errors.js"; @@ -167,7 +169,7 @@ function normalizeConnectionSummary( function normalizeProviderMetadata(metadata: Record | null | undefined): Record | null { if (!metadata) return null; - return redactEnvironmentCustomImageValue(metadata); + return redactEnvironmentCustomImageValue(withoutCleanupVerification(metadata) ?? {}); } function metadataRecord(metadata: Record | null | undefined): Record { @@ -647,7 +649,7 @@ export function environmentCustomImageService( return await markSessionStatus({ sessionId: session.id, status, - metadata: cancelled.metadata, + metadata: mergeSetupSessionMetadata(session.metadata, cancelled.metadata), failureReason: status === "failed" ? "Provider failed to cancel setup session." : null, }); } catch (error) { @@ -760,6 +762,22 @@ export function environmentCustomImageService( getSessionById, + async inspectSetupResource(session: typeof environmentCustomImageSetupSessions.$inferSelect): Promise { + if (!session.providerLeaseId || !session.finishedAt || session.connectionSecretRef !== null || + !["promoted", "cancelled", "timed_out", "failed"].includes(session.status)) return undefined; + const environment = await requireEnvironment(session.environmentId); + const provider = await resolveSetupProvider({ environment, storedProvider: session.provider, + storedRpcCompanyId: readSetupRpcCompanyId(session.metadata) }); + const recordedFingerprint = readString(session.metadata?.[SOURCE_ENVIRONMENT_CONFIG_FINGERPRINT_METADATA_KEY]); + if (!recordedFingerprint || fingerprintEnvironmentSandboxProviderConfig(provider.runtimeConfig) !== recordedFingerprint) return undefined; + if (!options.pluginWorkerManager?.getWorker(provider.pluginId)?.supportedMethods?.includes("environmentInspectLease")) return undefined; + return await options.pluginWorkerManager.call(provider.pluginId, "environmentInspectLease", { + driverKey: provider.provider, companyId: provider.rpcCompanyId, environmentId: environment.id, + issueId: null, config: provider.driverConfig, providerLeaseId: session.providerLeaseId, + leaseMetadata: session.metadata ?? {}, + }, 15_000); + }, + startSetupSession: async (input: { environmentId: string; templateId?: string | null; diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index c488b4c5912..a95d58cec6e 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1,6 +1,7 @@ import { allowLegacyShutdownWorkspaceCleanup, legacyShutdownWorkspaceResourceProtected } from "./legacy-shutdown-workspace-settlement.js"; import { nativeWorkspaceExportPendingForLease } from "./native-runtime/native-workspace-finalization-state.js"; import { beginIdleTrackedWork } from "./task-admission.js"; +import { historicalProviderCredentials } from "./historical-provider-credentials.js"; import { hasStopOnlyCleanup, prepareSandboxStopAndRetain, readStopOnlyCleanup, settleStopOnlyCleanup, stopOnlyCleanupKey } from "./sandbox-stop-and-retain.js"; import { JsonRpcCallError, readEnvironmentAcquisitionDiagnostic, readEnvironmentCreationCleanupError } from "@paperclipai/plugin-sdk"; import { preserveEnvironmentSyncOutErrorDiagnostic } from "./environment-sync-out-error.js"; @@ -32,6 +33,7 @@ import type { PluginEnvironmentAcquireLeaseParams, PluginEnvironmentExecuteResult, PluginEnvironmentLease, + PluginEnvironmentLeaseObservation, PluginEnvironmentRealizeWorkspaceResult, PluginEnvironmentSyncResult, PluginSyncOperation, @@ -592,6 +594,7 @@ export interface EnvironmentDriverRunnerIngressInput export interface EnvironmentRuntimeDriver { readonly driver: string; + inspectLeaseResource?(input: { environment: Environment | null; lease: EnvironmentLease }): Promise; acquireRunLease(input: EnvironmentDriverAcquireInput): Promise; releaseRunLease(input: EnvironmentDriverReleaseInput): Promise; resumeRunLease?(input: EnvironmentDriverLeaseInput): Promise; @@ -2610,6 +2613,28 @@ function createSandboxEnvironmentDriver( }); }, + async inspectLeaseResource({ lease }) { + const provider = lease.provider; + const config = sandboxConfigFromLeaseMetadataLoose(lease); + if (!provider || isBuiltinSandboxProvider(provider) || !config || config.provider !== provider || + !lease.providerLeaseId || !lease.environmentId || !pluginWorkerManager) return undefined; + const pluginId = readString(lease.metadata?.pluginId); + // Historical observations must use the recorded plugin and credentials. + // A different installation cannot establish that this allocation is gone. + if (!pluginId) return undefined; + const resolved = await resolveSandboxProviderPlugin({ provider, pluginId }); + if (resolved.state !== "running" || !pluginWorkerVerifiesLifecycleMethod(pluginId, "environmentInspectLease")) return undefined; + const resolvedConfig = await historicalProviderCredentials(db, lease.companyId, config, + resolved.resolved.driver.configSchema as Record | undefined, + { issueId: lease.issueId, heartbeatRunId: lease.heartbeatRunId }); + if (!resolvedConfig) return undefined; + return await pluginWorkerManager.call(pluginId, "environmentInspectLease", { + driverKey: provider, companyId: lease.companyId, environmentId: lease.environmentId, + issueId: lease.issueId, config: stripSandboxProviderEnvelope(resolvedConfig as SandboxEnvironmentConfig), + providerLeaseId: lease.providerLeaseId, leaseMetadata: lease.metadata ?? {}, + }, 15_000); + }, + async retryPendingSandboxTeardown(input) { const exportResume = hasStopOnlyCleanup(input.lease); if (!exportResume && await nativeWorkspaceExportPendingForLease(db, input.lease)) { @@ -4235,6 +4260,13 @@ export function environmentRuntimeService( // record, so a deleted or foreign-bound environment never strands it. The // caller owns the lease release; this dispatcher only runs the provider // teardown and throws when the driver teardown throws. + async inspectLeaseResource(leaseId: string): Promise { + const lease = await environmentsSvc.getLeaseById(leaseId); + if (!lease || lease.status !== "failed" || lease.leasePolicy !== "ephemeral" || !lease.providerLeaseId) return undefined; + const environment = lease.environmentId ? await environmentsSvc.getById(lease.environmentId) : null; + return await getDriver(getLeaseDriverKey(lease, environment))?.inspectLeaseResource?.({ environment, lease }); + }, + async retryPendingSandboxTeardown(input: { environment: Environment | null; lease: EnvironmentLease; diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index b8922aa61d7..0ac6c69fcbc 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1,3 +1,4 @@ +import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { randomUUID } from "node:crypto"; import { SANDBOX_PLATFORM_PROBE_METADATA_KEY } from "./environment-custom-image-runtime.js"; @@ -1463,7 +1464,7 @@ export function environmentService(db: Db) { releasedAt: null, failureReason: null, cleanupStatus: null, - metadata: input.metadata ?? null, + metadata: withoutCleanupVerification(input.metadata), createdAt: now, updatedAt: now, }; @@ -1584,7 +1585,7 @@ export function environmentService(db: Db) { expiresAt: input.expiresAt ?? null, failureReason: null, cleanupStatus: null, - metadata: input.metadata ?? null, + metadata: withoutCleanupVerification(input.metadata), updatedAt: now, }) .where( @@ -1739,7 +1740,7 @@ export function environmentService(db: Db) { releasedAt: now, failureReason: input.failureReason, cleanupStatus: "failed", - metadata: input.metadata ?? null, + metadata: withoutCleanupVerification(input.metadata), createdAt: now, updatedAt: now, }) @@ -1759,7 +1760,7 @@ export function environmentService(db: Db) { const row = await db .update(environmentLeases) .set({ - metadata, + metadata: withoutCleanupVerification(metadata), lastUsedAt: new Date(), updatedAt: new Date(), }) diff --git a/server/src/services/historical-cleanup-verification.ts b/server/src/services/historical-cleanup-verification.ts new file mode 100644 index 00000000000..405c2ae69fa --- /dev/null +++ b/server/src/services/historical-cleanup-verification.ts @@ -0,0 +1,150 @@ +import { and, eq, inArray, sql } from "drizzle-orm"; +import { environmentCustomImageSetupSessions, environmentLeases, type Db } from "@paperclipai/db"; +import { beginIdleTrackedWork } from "./task-admission.js"; +import { finishedIdleRun } from "./idle-work-proof.js"; +import { CLEANUP_ROW_STAMP_FORMAT, CLEANUP_VERIFICATION_KEY, CLEANUP_VERIFICATION_SCHEMA, cleanupRowStamp, verifiedResourceCleanup } from "./cleanup-verification-proof.js"; + +type Lease = typeof environmentLeases.$inferSelect; +type Setup = typeof environmentCustomImageSetupSessions.$inferSelect; +type Inspectors = { + inspectLease(lease: Lease): Promise; + inspectSetup(setup: Setup): Promise; +}; +type Result = { checked: number; verified: number; deferred: number; failed: number }; + +/** Read provider state only. Preserve failures and retained data; observing an + * allocation never starts or destroys it. Older workers simply defer proof. */ +export function historicalCleanupVerificationService(db: Db, inspectors: Inspectors) { + let inFlight: Promise | undefined; + let leaseCursor = ""; + let setupCursor = ""; + async function run(): Promise { + const finish = beginIdleTrackedWork(); + const result: Result = { checked: 0, verified: 0, deferred: 0, failed: 0 }; + try { + const leases = await db.select({ row: environmentLeases, stamp: sql`${sql.raw(cleanupRowStamp('environment_leases'))}` }) + .from(environmentLeases).where(and( + eq(environmentLeases.status, "failed"), eq(environmentLeases.leasePolicy, "ephemeral"), + eq(environmentLeases.cleanupStatus, "success"), + sql`${environmentLeases.id}::text > ${leaseCursor}`, + sql`${environmentLeases.releasedAt} IS NOT NULL AND ${environmentLeases.environmentId} IS NOT NULL`, + sql`${environmentLeases.provider} IS NOT NULL AND ${environmentLeases.provider} <> 'local' + AND ${environmentLeases.providerLeaseId} IS NOT NULL`, + sql`COALESCE(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') = 'false' + AND NOT (COALESCE(${environmentLeases.metadata}, '{}'::jsonb) ?| ARRAY['nativeWorkspaceExportResume', 'sandboxStopAndRetain'])`, + sql`(${sql.raw(verifiedResourceCleanup('environment_leases', 'lease'))}) IS NOT TRUE`, + sql`(${environmentLeases.heartbeatRunId} IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs historical_run + WHERE historical_run.id = ${environmentLeases.heartbeatRunId} AND historical_run.company_id = ${environmentLeases.companyId} + AND ${sql.raw(finishedIdleRun('historical_run'))}))`, + sql`NOT EXISTS (SELECT 1 FROM environment_leases other_owner + WHERE other_owner.id <> ${environmentLeases.id} AND other_owner.provider = ${environmentLeases.provider} + AND other_owner.provider_lease_id = ${environmentLeases.providerLeaseId} + AND other_owner.status IN ('active', 'retained', 'pending_cleanup'))`, + sql`NOT EXISTS (SELECT 1 FROM environment_custom_image_setup_sessions setup_owner + WHERE setup_owner.provider = ${environmentLeases.provider} AND setup_owner.provider_lease_id = ${environmentLeases.providerLeaseId} + AND (setup_owner.finished_at IS NULL OR setup_owner.connection_secret_ref IS NOT NULL))`, + )).orderBy(environmentLeases.id).limit(16); + const setups = await db.select({ row: environmentCustomImageSetupSessions, + stamp: sql`${sql.raw(cleanupRowStamp('environment_custom_image_setup_sessions'))}` }) + .from(environmentCustomImageSetupSessions).where(and( + inArray(environmentCustomImageSetupSessions.status, ["promoted", "cancelled", "timed_out", "failed"]), + sql`${environmentCustomImageSetupSessions.id}::text > ${setupCursor}`, + sql`${environmentCustomImageSetupSessions.finishedAt} IS NOT NULL + AND ${environmentCustomImageSetupSessions.connectionSecretRef} IS NULL + AND ${environmentCustomImageSetupSessions.providerLeaseId} IS NOT NULL`, + sql`(${sql.raw(verifiedResourceCleanup('environment_custom_image_setup_sessions', 'image_setup'))}) IS NOT TRUE`, + sql`NOT EXISTS (SELECT 1 FROM environment_leases lease_owner + WHERE lease_owner.provider = ${environmentCustomImageSetupSessions.provider} + AND lease_owner.provider_lease_id = ${environmentCustomImageSetupSessions.providerLeaseId} + AND lease_owner.status NOT IN ('released', 'expired', 'failed'))`, + )).orderBy(environmentCustomImageSetupSessions.id).limit(16); + leaseCursor = leases.length === 16 ? leases.at(-1)!.row.id : ""; + setupCursor = setups.length === 16 ? setups.at(-1)!.row.id : ""; + // Keep provider concurrency bounded. Each operation is an observation; + // every accepted receipt must also win the row-version comparison below. + const jobs = [ + ...leases.map(snapshot => async () => inspect("lease", snapshot.row, snapshot.stamp)), + ...setups.map(snapshot => async () => inspect("image_setup", snapshot.row, snapshot.stamp)), + ]; + let next = 0; + await Promise.all(Array.from({ length: Math.min(2, jobs.length) }, async () => { + for (;;) { const index = next++; if (index >= jobs.length) return; await jobs[index]!(); } + })); + return result; + } finally { finish(); } + + async function inspect(kind: "lease" | "image_setup", row: Lease | Setup, stamp: string) { + result.checked += 1; + try { + const observation = await (kind === "lease" ? inspectors.inspectLease(row as Lease) : inspectors.inspectSetup(row as Setup)) as { + providerLeaseId?: unknown; state?: unknown; + } | null | undefined; + if (observation?.state !== "destroyed" || !row.providerLeaseId || observation.providerLeaseId !== row.providerLeaseId) { + result.deferred += 1; return; + } + const now = new Date(); + const companyId = kind === "lease" ? (row as Lease).companyId + : typeof row.metadata?.setupRpcCompanyId === "string" && row.metadata.setupRpcCompanyId.trim() + ? row.metadata.setupRpcCompanyId.trim() : "instance"; + const proof = { schema: CLEANUP_VERIFICATION_SCHEMA, kind, resourceId: row.id, companyId, + environmentId: row.environmentId, provider: row.provider, providerLeaseId: row.providerLeaseId, + heartbeatRunId: kind === "lease" ? (row as Lease).heartbeatRunId : null, + state: "destroyed", confirmedAt: now.toISOString() }; + const metadata = sql`COALESCE(${kind === "lease" ? environmentLeases.metadata : environmentCustomImageSetupSessions.metadata}, '{}'::jsonb) + || jsonb_build_object(${CLEANUP_VERIFICATION_KEY}::text, ${JSON.stringify(proof)}::jsonb + || jsonb_build_object('rowUpdatedAt', to_char(${now.toISOString()}::timestamptz AT TIME ZONE 'UTC', ${CLEANUP_ROW_STAMP_FORMAT})))`; + let updated: Array<{ id: string }>; + if (kind === "lease") { + const lease = row as Lease; + updated = await db.update(environmentLeases).set({ metadata, updatedAt: now }).where(and( + eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId), + eq(environmentLeases.status, "failed"), eq(environmentLeases.leasePolicy, "ephemeral"), + eq(environmentLeases.cleanupStatus, "success"), + sql`${environmentLeases.provider} IS NOT DISTINCT FROM ${lease.provider} + AND ${environmentLeases.providerLeaseId} IS NOT DISTINCT FROM ${lease.providerLeaseId} + AND ${environmentLeases.environmentId} IS NOT DISTINCT FROM ${lease.environmentId}::uuid + AND ${environmentLeases.heartbeatRunId} IS NOT DISTINCT FROM ${lease.heartbeatRunId}::uuid + AND ${environmentLeases.releasedAt} IS NOT NULL`, + sql`(${environmentLeases.heartbeatRunId} IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs verify_run + WHERE verify_run.id = ${environmentLeases.heartbeatRunId} AND verify_run.company_id = ${environmentLeases.companyId} + AND ${sql.raw(finishedIdleRun('verify_run'))}))`, + sql`${sql.raw(cleanupRowStamp('environment_leases'))} = ${stamp} + AND ${environmentLeases.metadata} IS NOT DISTINCT FROM ${(lease.metadata === null ? null : JSON.stringify(lease.metadata))}::jsonb`, + sql`NOT EXISTS (SELECT 1 FROM environment_leases other_owner + WHERE other_owner.id <> ${environmentLeases.id} AND other_owner.provider = ${environmentLeases.provider} + AND other_owner.provider_lease_id = ${environmentLeases.providerLeaseId} + AND other_owner.status IN ('active', 'retained', 'pending_cleanup'))`, + sql`NOT EXISTS (SELECT 1 FROM environment_custom_image_setup_sessions setup_owner + WHERE setup_owner.provider = ${environmentLeases.provider} AND setup_owner.provider_lease_id = ${environmentLeases.providerLeaseId} + AND (setup_owner.finished_at IS NULL OR setup_owner.connection_secret_ref IS NOT NULL))`, + )).returning({ id: environmentLeases.id }); + } else { + const setup = row as Setup; + updated = await db.update(environmentCustomImageSetupSessions).set({ metadata, updatedAt: now }).where(and( + eq(environmentCustomImageSetupSessions.id, setup.id), eq(environmentCustomImageSetupSessions.status, setup.status), + sql`${environmentCustomImageSetupSessions.provider} IS NOT DISTINCT FROM ${setup.provider} + AND ${environmentCustomImageSetupSessions.providerLeaseId} IS NOT DISTINCT FROM ${setup.providerLeaseId} + AND ${environmentCustomImageSetupSessions.environmentId} = ${setup.environmentId}::uuid + AND ${environmentCustomImageSetupSessions.finishedAt} IS NOT NULL`, + sql`NOT EXISTS (SELECT 1 FROM environment_leases lease_owner + WHERE lease_owner.provider = ${environmentCustomImageSetupSessions.provider} + AND lease_owner.provider_lease_id = ${environmentCustomImageSetupSessions.providerLeaseId} + AND lease_owner.status NOT IN ('released', 'expired', 'failed'))`, + sql`${sql.raw(cleanupRowStamp('environment_custom_image_setup_sessions'))} = ${stamp} + AND ${environmentCustomImageSetupSessions.metadata} IS NOT DISTINCT FROM ${(setup.metadata === null ? null : JSON.stringify(setup.metadata))}::jsonb + AND ${environmentCustomImageSetupSessions.connectionSecretRef} IS NULL`, + )).returning({ id: environmentCustomImageSetupSessions.id }); + } + if (updated.length === 1) result.verified += 1; + else result.deferred += 1; + } catch { + // Provider exceptions may contain credentials. Keep only a count. + result.failed += 1; + } + } + } + return { sweep(): Promise { + if (!inFlight) inFlight = run().finally(() => { inFlight = undefined; }); + return inFlight; + } }; +} diff --git a/server/src/services/historical-provider-credentials.ts b/server/src/services/historical-provider-credentials.ts new file mode 100644 index 00000000000..d36fe4009b7 --- /dev/null +++ b/server/src/services/historical-provider-credentials.ts @@ -0,0 +1,40 @@ +import { createHash } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { companySecrets, companySecretVersions, type Db } from "@paperclipai/db"; +import type { SandboxEnvironmentConfig } from "@paperclipai/shared"; +import { collectSecretRefPaths, isUuidSecretRef, parseSecretRefBindingObject, readConfigValueAtPath, writeConfigValueAtPath } from "./json-schema-secret-refs.js"; +import { secretService } from "./secrets.js"; + +/** A 404 under replacement credentials says nothing about the old resource. + * Legacy leases did not record a secret-version manifest. Only an unrotated + * initial version whose resolved bytes still match its immutable hash can + * establish their credential identity. Other histories need explicit evidence. + * Normal cleanup resolution and run-continuation rules remain unchanged. */ +export async function historicalProviderCredentials(db: Db, companyId: string, + config: SandboxEnvironmentConfig, schema: Record | null | undefined, + context: { issueId: string | null; heartbeatRunId: string | null }, +): Promise { + let resolved = { ...config } as Record; + for (const path of collectSecretRefPaths(schema)) { + const raw = readConfigValueAtPath(resolved, path); + const binding = parseSecretRefBindingObject(raw); + if (binding && binding.version !== "latest" && binding.version !== 1) return undefined; + const ref = binding?.secretId ?? (typeof raw === "string" ? raw.trim() : null); + if (!ref || !isUuidSecretRef(ref)) continue; + const [version] = await db.select({ hash: companySecretVersions.valueSha256 }) + .from(companySecrets).innerJoin(companySecretVersions, and( + eq(companySecretVersions.secretId, companySecrets.id), eq(companySecretVersions.version, 1))) + .where(and(eq(companySecrets.id, ref), eq(companySecrets.companyId, companyId), + eq(companySecrets.latestVersion, 1), eq(companySecrets.status, "active"), + sql`${companySecretVersions.revokedAt} IS NULL`, + sql`NOT EXISTS (SELECT 1 FROM company_secret_versions other_version + WHERE other_version.secret_id = ${companySecrets.id} AND other_version.version <> 1)`)); + if (!version) return undefined; + const value = await secretService(db).resolveSecretValueForSandboxCleanup(companyId, ref, 1, { ...context, configPath: path }); + if (createHash("sha256").update(value).digest("hex") !== version.hash) return undefined; + resolved = writeConfigValueAtPath(resolved, path, value); + } + if (resolved.provider === "daytona" && (typeof resolved.apiKey !== "string" || + !resolved.apiKey.trim() || isUuidSecretRef(resolved.apiKey.trim()))) return undefined; + return resolved as SandboxEnvironmentConfig; +} diff --git a/server/src/services/idle-issue-schedule.ts b/server/src/services/idle-issue-schedule.ts index 431ea72f2fb..e3944d75a0c 100644 --- a/server/src/services/idle-issue-schedule.ts +++ b/server/src/services/idle-issue-schedule.ts @@ -4,6 +4,7 @@ import type { Db } from "@paperclipai/db"; import { AGENT_STATUSES, issueExecutionMonitorPolicySchema, PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; import type { ScheduledWake } from "./idle-scheduled-work.js"; import { onboardingFirstTaskAwaitingUserSql } from "./recovery/onboarding-first-task-wait.js"; +import { responseDrivenIssueWait } from "./idle-work-proof.js"; const processGeneration = randomUUID(); const limit = 10_000; @@ -15,6 +16,7 @@ const record = (value: unknown): Record | null => export async function readIdleIssueSchedule(tx: Pick, ownerId: string, now: number): Promise<{ blocked: boolean; scheduledWake?: ScheduledWake; + generation?: string; }> { const rows = await tx.execute<{ id: string; company_id: string; status: string; updated_at: Date | string; @@ -25,13 +27,18 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st monitor_wake_requested_at: Date | string | null; monitor_attempt_count: number; company_status: string; company_updated: Date | string; agent_status: string | null; agent_updated: Date | string | null; scoped: boolean; unsafe_dependencies: boolean; dependency_ready: boolean; - untouched_onboarding: boolean; + untouched_onboarding: boolean; response_driven_wait: boolean; + response_wait_generation: unknown; }>(sql`SELECT i.id, i.company_id, i.status, i.updated_at, i.assignee_agent_id, i.assignee_user_id, i.checkout_run_id, i.execution_run_id, i.execution_policy, i.execution_state, i.unblock_descriptor, i.conversation_agent_id, i.monitor_next_check_at, i.monitor_wake_requested_at, i.monitor_attempt_count, c.status AS company_status, c.updated_at AS company_updated, a.status AS agent_status, a.updated_at AS agent_updated, (i.assignee_agent_id IS NULL OR a.id IS NOT NULL) AS scoped, + (${sql.raw(responseDrivenIssueWait('i'))}) AS response_driven_wait, + (SELECT jsonb_agg(jsonb_build_array(w.id, w.updated_at, w.kind, w.status, w.source_run_id, + w.effective_resolver_policy, w.continuation_policy, w.addressee_agent_id, w.addressee_user_id) ORDER BY w.id) + FROM issue_thread_interactions w WHERE w.issue_id = i.id AND w.status = 'pending') AS response_wait_generation, (${onboardingFirstTaskAwaitingUserSql(sql`i.company_id`, sql`i.id`, sql`i.origin_kind`)} AND NOT EXISTS (SELECT 1 FROM heartbeat_runs h WHERE h.company_id = i.company_id AND (h.context_snapshot ->> 'issueId' = i.id::text OR h.context_snapshot ->> 'taskId' = i.id::text))) AS untouched_onboarding, @@ -76,17 +83,22 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st || (row.status === "todo" && !row.assignee_agent_id) || (row.status === "todo" && row.assignee_agent_id && row.agent_status === "idle" && row.untouched_onboarding === true) || (row.status === "blocked" && (row.unblock_descriptor === null || unblock?.owner === "board")) + || (["in_progress", "in_review"].includes(row.status) && row.response_driven_wait === true) || Boolean(row.assignee_user_id && !row.assignee_agent_id); if (!passive) return { blocked: true }; } - if (!Number.isFinite(earliest)) return { blocked: false }; + const generation = createHash("sha256").update(JSON.stringify([processGeneration, ownerId, rows])).digest("hex"); + if (!Number.isFinite(earliest)) return { blocked: false, generation }; return { blocked: false, scheduledWake: { version: 1, kind: "routines_and_issue_monitors", - generation: createHash("sha256").update(JSON.stringify([processGeneration, ownerId, rows])).digest("hex"), + generation, nextWakeAt: new Date(earliest).toISOString() } }; } -export function combineIdleSchedules(routine?: ScheduledWake, issue?: ScheduledWake): ScheduledWake | undefined { - if (!issue) return routine; +export function combineIdleSchedules(routine?: ScheduledWake, issue?: ScheduledWake, issueGeneration?: string): ScheduledWake | undefined { + // A passive issue can change even when only a routine owns the next alarm. + // Bind that waiting-state proof into the routine generation as well. + if (!issue) return routine && issueGeneration ? { ...routine, + generation: createHash("sha256").update(JSON.stringify([routine, issueGeneration])).digest("hex") } : routine; if (!routine) return issue; return { version: 1, kind: "routines_and_issue_monitors", generation: createHash("sha256").update(JSON.stringify([routine, issue])).digest("hex"), diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index c5dfb630490..579cbf39da9 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -2,8 +2,10 @@ import { 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"; +import { connectionDeliveryWaitingForResponse, finishedIdleRun as finishedRun, finishedIdleWake, passiveBoardRecovery } from "./idle-work-proof.js"; import type { Db } from "@paperclipai/db"; import { sql } from "drizzle-orm"; +import { verifiedResourceCleanup } from "./cleanup-verification-proof.js"; export interface IdleSleepDrainStatus { ownerId?: string; @@ -25,16 +27,6 @@ export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number }) pluginIds: string[]; }>; -// Retry promotion keeps its due date as history on the run. Only a terminal -// run finished at or after that date has consumed the schedule. Newer retry -// dates, unfinished runs and accounting work still require the instance. -function finishedRun(alias: string): string { - return `${alias}.status IN ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted') - AND ${alias}.finished_at IS NOT NULL - AND (${alias}.scheduled_retry_at IS NULL OR ${alias}.scheduled_retry_at <= ${alias}.finished_at) - AND ${alias}.cost_accounting_pending = false`; -} - // A failed turn may have successfully torn down its ephemeral allocation. A // cleanup label alone is not provider termination proof. Require the exact // host-owned receipt and finished run, and retain all cleanup/export intents. @@ -42,21 +34,28 @@ function finishedRun(alias: string): string { function finishedFailedLease(alias: string): string { const receipt = `${alias}.metadata->'remoteExecutionTermination'`; return `${alias}.status = 'failed' AND ${alias}.lease_policy = 'ephemeral' - AND ${alias}.cleanup_status = 'success' AND ${alias}.released_at IS NOT NULL + AND ${alias}.released_at IS NOT NULL AND COALESCE(${alias}.metadata->>'pendingCleanupInFlight', 'false') = 'false' AND NOT (COALESCE(${alias}.metadata, '{}'::jsonb) ?| ARRAY['nativeWorkspaceExportResume', 'sandboxStopAndRetain']) - AND ${alias}.provider IS NOT NULL AND ${alias}.provider <> 'local' - AND ${alias}.provider_lease_id IS NOT NULL AND ${alias}.provider_lease_id <> '' - AND ${receipt}->>'schema' = 'paperclip.remote-termination.v1' - AND ${receipt}->>'companyId' = ${alias}.company_id::text - AND ${receipt}->>'runId' = ${alias}.heartbeat_run_id::text - AND ${receipt}->>'leaseId' = ${alias}.id::text - AND ${receipt}->>'provider' = ${alias}.provider - AND ${receipt}->>'providerLeaseId' = ${alias}.provider_lease_id - AND ${receipt}->>'state' IN ('stopped', 'destroyed') - AND EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = ${alias}.heartbeat_run_id + AND (( ${alias}.provider = 'local' AND ${alias}.provider_lease_id IS NULL + AND ${alias}.metadata->>'driver' = 'local' + AND ${receipt} IS NULL + AND EXISTS (SELECT 1 FROM environments local_environment + WHERE local_environment.id = ${alias}.environment_id AND local_environment.driver = 'local')) + OR (${alias}.cleanup_status = 'success' AND ${alias}.provider IS NOT NULL AND ${alias}.provider <> 'local' + AND ${alias}.provider_lease_id IS NOT NULL AND ${alias}.provider_lease_id <> '' + AND ((${receipt}->>'schema' = 'paperclip.remote-termination.v1' + AND ${receipt}->>'companyId' = ${alias}.company_id::text + AND ${receipt}->>'runId' = ${alias}.heartbeat_run_id::text + AND ${receipt}->>'leaseId' = ${alias}.id::text + AND ${receipt}->>'provider' = ${alias}.provider + AND ${receipt}->>'providerLeaseId' = ${alias}.provider_lease_id + AND ${receipt}->>'state' IN ('stopped', 'destroyed')) + OR (${verifiedResourceCleanup(alias, 'lease')})))) + AND ((${alias}.heartbeat_run_id IS NULL AND ${alias}.provider <> 'local' + AND (${verifiedResourceCleanup(alias, 'lease')})) OR EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = ${alias}.heartbeat_run_id AND r.company_id = ${alias}.company_id - AND ${finishedRun("r")})`; + AND ${finishedRun("r")}))`; } // These are deliberately conservative. Completed agent runs and workspace @@ -128,17 +127,45 @@ const WORK_CHECKS = [ ) IS NOT TRUE`, `SELECT 1 FROM environment_leases l WHERE status NOT IN ('released', 'expired') AND (${finishedFailedLease("l")}) IS NOT TRUE`, + `SELECT 1 FROM environment_custom_image_setup_sessions s WHERE + (s.status IN ('promoted', 'cancelled', 'timed_out', 'failed') AND s.finished_at IS NOT NULL + AND s.connection_secret_ref IS NULL AND (${verifiedResourceCleanup('s', 'image_setup')})) IS NOT TRUE`, + // An unconfigured Slack draft has no provider identity or inbound runtime. + // Setup still wakes through normal browser admission. Retain every enabled, + // previously active, partially connected, foreign, or unsupported endpoint. + `SELECT 1 FROM chat_endpoints endpoint WHERE (endpoint.provider = 'slack' + AND endpoint.status = 'draft' AND endpoint.setup->>'step' = 'provider_setup' + AND (endpoint.setup - ARRAY['step', 'command']) = '{}'::jsonb + AND (endpoint.setup->'command' IS NULL OR endpoint.setup->'command' = 'null'::jsonb + OR jsonb_typeof(endpoint.setup->'command') = 'string') + AND endpoint.activated_at IS NULL AND endpoint.provider_account_id IS NULL + AND endpoint.bot_external_id IS NULL AND endpoint.bot_username IS NULL + AND endpoint.last_event_at IS NULL AND endpoint.last_publication_at IS NULL + AND endpoint.health_message IS NULL AND endpoint.last_error IS NULL + AND NOT EXISTS (SELECT 1 FROM chat_slack_registrations registration WHERE registration.endpoint_id = endpoint.id) + AND EXISTS (SELECT 1 FROM tool_connections draft_connection + WHERE draft_connection.id = endpoint.connection_id AND draft_connection.company_id = endpoint.company_id + AND draft_connection.connection_purpose = 'channel' AND draft_connection.transport = 'chat_sdk' + AND draft_connection.status = 'draft' AND draft_connection.enabled = false)) IS NOT TRUE`, `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`, `SELECT 1 FROM routine_runs WHERE status NOT IN ('completed', 'failed', 'skipped', 'cancelled')`, `SELECT 1 FROM plugin_jobs WHERE status <> 'paused'`, `SELECT 1 FROM plugin_job_runs WHERE status NOT IN ('succeeded', 'failed', 'cancelled')`, `SELECT 1 FROM plugin_webhook_deliveries WHERE status NOT IN ('succeeded', 'failed')`, - // Saved AI credentials have no independent worker. Login sessions, accepted - // quota/refresh work and recovery journals have their own checks. Other - // integrations still need an ingress/scheduler wake contract. - `SELECT 1 FROM tool_connections WHERE status <> 'archived' - AND NOT (status = 'active' AND connection_purpose = 'ai' AND transport = 'runtime_auth')`, - `SELECT 1 FROM issue_recovery_actions WHERE status NOT IN ('resolved', 'cancelled')`, + // Saved outbound MCP configuration and disabled drafts have no independent + // listener. Keep local processes, channels and GitHub webhook bindings awake; + // accepted calls, credential recovery and delivery work have separate fences. + `SELECT 1 FROM tool_connections c WHERE c.status <> 'archived' AND ( + (c.status = 'draft' AND c.enabled = false) + OR (c.status = 'active' AND c.connection_purpose = 'ai' AND c.transport = 'runtime_auth') + OR (c.status = 'active' AND c.connection_purpose = 'tool' AND c.transport = 'mcp_remote' + AND NOT EXISTS (SELECT 1 FROM connection_grants g WHERE g.connection_id = c.id + AND (g.company_id <> c.company_id OR g.provider_tenant->'github' IS NOT NULL))) + ) IS NOT TRUE`, + // An explicit board escalation has no automatic retry. Retain the visible + // action and all execution/cleanup checks while the board decides what to do. + `SELECT 1 FROM issue_recovery_actions a WHERE a.status NOT IN ('resolved', 'cancelled') + AND (${passiveBoardRecovery('a')}) IS NOT TRUE`, `SELECT 1 FROM status_cards WHERE archived_at IS NULL`, `SELECT 1 FROM external_objects WHERE NOT is_terminal OR next_refresh_at IS NOT NULL`, // Login history keeps its provider reference after teardown. A terminal @@ -186,17 +213,30 @@ const WORK_CHECKS = [ SELECT 1 FROM heartbeat_runs r WHERE r.id = w.run_id AND r.company_id = w.company_id AND r.agent_id = w.agent_id AND ${finishedRun("r")})))))`, + // Cloud owns inbound event durability. A processed receipt is idempotency + // history; any downstream run, wake or delivery is still checked separately. + `SELECT 1 FROM connection_event_deliveries WHERE status <> 'processed' OR processed_at IS NULL`, + // A delivered outbox row is history after its exact accepted wake finishes. + // A system-retired connection card may also be parked behind a still-pending + // response. Keep the outbox intact: the normal reply path resumes delivery. + `SELECT 1 FROM connection_intent_deliveries d WHERE + (d.delivered_at IS NULL AND (${connectionDeliveryWaitingForResponse('d')}) IS NOT TRUE) + OR EXISTS (SELECT 1 FROM agent_wakeup_requests connection_wake + WHERE connection_wake.company_id = d.company_id AND connection_wake.idempotency_key IN ( + 'connection-intent:' || d.interaction_id::text || ':accepted', + 'connection-intent:' || d.interaction_id::text || ':rejected', + 'connection-intent:' || d.interaction_id::text || ':expired') + AND (${finishedIdleWake('connection_wake')}) IS NOT TRUE)`, // These less common work sources fail closed on any retained state. Their // terminal-state exceptions can be added with tests for the owning service. ...[ - "chat_endpoints", "chat_deliveries", "chat_publications", "chat_actions", - "chat_completion_deliveries", "connection_event_deliveries", - "connection_intent_deliveries", "tool_action_requests", "tool_action_deliveries", + "chat_deliveries", "chat_publications", "chat_actions", + "chat_completion_deliveries", "tool_action_requests", "tool_action_deliveries", "workspace_runtime_services", "execution_workspace_runtime_leases", "pipeline_automation_executions", "native_run_finalizations", "company_transfer_runs", "decisions", "decision_effect_executions", "decision_archive_notification_outbox", "browser_use_sessions", "browser_use_runs", - "browser_use_browsers", "environment_custom_image_setup_sessions", "feedback_exports", + "browser_use_browsers", "feedback_exports", "company_secret_proposals", "mcp_oauth_grants", "mcp_mutation_receipts", "mcp_event_subscriptions", "mcp_event_deliveries", "mcp_attachment_uploads", "dot_agent_bindings", @@ -266,7 +306,7 @@ export async function readIdleSleepSafety( if (!scheduledRoutines || !ownerId || ownerId !== before.ownerId) return undefined; const schedule = await readIdleIssueSchedule(tx, ownerId, now()); if (schedule.blocked) return true; - scheduledWake = combineIdleSchedules(scheduledWake, schedule.scheduledWake); + scheduledWake = combineIdleSchedules(scheduledWake, schedule.scheduledWake, schedule.generation); } return false; }, { isolationLevel: "repeatable read", accessMode: "read only" }); diff --git a/server/src/services/idle-work-proof.ts b/server/src/services/idle-work-proof.ts new file mode 100644 index 00000000000..38ef426696d --- /dev/null +++ b/server/src/services/idle-work-proof.ts @@ -0,0 +1,96 @@ +/** Fixed SQL aliases only. These predicates do not accept request input. */ +export function finishedIdleRun(alias: string): string { + return `${alias}.status IN ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted') + AND ${alias}.finished_at IS NOT NULL + AND (${alias}.scheduled_retry_at IS NULL OR ${alias}.scheduled_retry_at <= ${alias}.finished_at) + AND ${alias}.cost_accounting_pending = false`; +} + +export function finishedIdleWake(alias: string): string { + return `${alias}.finished_at IS NOT NULL AND ( + ${alias}.status IN ('completed', 'failed', 'cancelled', 'skipped', 'timed_out') + OR (${alias}.status = 'coalesced' AND EXISTS ( + SELECT 1 FROM heartbeat_runs wake_target WHERE wake_target.id = ${alias}.run_id + AND wake_target.company_id = ${alias}.company_id AND wake_target.agent_id = ${alias}.agent_id + AND ${finishedIdleRun('wake_target')})))`; +} + +export function connectionDeliveryWaitingForResponse(delivery: string): string { + return `EXISTS (SELECT 1 FROM issue_thread_interactions waiting_delivery + JOIN issues waiting_issue ON waiting_issue.id = waiting_delivery.issue_id + AND waiting_issue.company_id = waiting_delivery.company_id + WHERE waiting_delivery.id = ${delivery}.interaction_id AND waiting_delivery.company_id = ${delivery}.company_id + AND waiting_delivery.kind = 'connection_intent' AND waiting_delivery.status = 'expired' + AND waiting_delivery.resolved_at IS NOT NULL AND waiting_delivery.result->>'outcome' = 'expired' + AND jsonb_typeof(waiting_delivery.result->'connectionId') = 'string' + AND waiting_issue.status IN ('in_progress', 'in_review') + AND waiting_issue.assignee_agent_id::text = waiting_delivery.payload->>'requestingAgentId' + AND (${responseDrivenIssueWait('waiting_issue')}))`; +} + +// Questions and confirmations advance on a response, not on a timer. Preserve +// the audience and continuation policy. A producer which still owns execution +// or accounting cannot certify a passive wait. +function passiveResponse(alias: string, issue: string): string { + return `${alias}.company_id = ${issue}.company_id + AND ${alias}.status = 'pending' AND ${alias}.resolved_at IS NULL + AND (${alias}.kind IN ('ask_user_questions', 'request_confirmation', 'request_checkbox_confirmation', 'suggest_tasks') + OR (${alias}.kind = 'connection_intent' AND ${alias}.effective_resolver_policy = 'human_only' + AND ${alias}.addressee_user_id IS NOT NULL AND ${issue}.assignee_user_id IS NULL + AND ${alias}.payload->>'version' = '1' + AND ${alias}.payload->>'requestingAgentId' = ${issue}.assignee_agent_id::text + AND length(${alias}.payload->>'serviceSlug') > 0 + AND (${alias}.payload->>'purpose' IS NULL OR ${alias}.payload->>'purpose' IN ('ai', 'channel')) + -- Existing access can automatically retire a setup card. In its + -- absence only the addressed human can finish this request. A broad + -- purpose match deliberately keeps unfamiliar credential paths awake. + AND NOT EXISTS (SELECT 1 FROM tool_connections response_connection + WHERE response_connection.company_id = ${issue}.company_id + AND response_connection.status = 'active' AND response_connection.enabled + AND response_connection.connection_purpose = COALESCE(${alias}.payload->>'purpose', 'tool')))) + -- Merge confirmations can advance through GitHub polling, without a + -- response. Its recognizer requires merge intent in these four fields; + -- conservatively retain those cards until polling has a durable wake. + AND (${alias}.kind <> 'request_confirmation' OR concat_ws(' ', ${alias}.title, ${alias}.summary, + ${alias}.payload->>'prompt', ${alias}.payload->>'acceptLabel') !~* 'merge') + AND ${alias}.effective_resolver_policy IN ('anyone', 'not_creator', 'human_only') + AND ${alias}.continuation_policy IN ('wake_assignee', 'none') + AND (${alias}.source_run_id IS NULL OR EXISTS ( + SELECT 1 FROM heartbeat_runs response_source WHERE response_source.id = ${alias}.source_run_id + AND response_source.company_id = ${issue}.company_id AND ${finishedIdleRun('response_source')}))`; +} + +export function responseDrivenIssueWait(issue: string): string { + return `EXISTS (SELECT 1 FROM issue_thread_interactions response_wait + WHERE response_wait.issue_id = ${issue}.id AND ${passiveResponse('response_wait', issue)}) + AND NOT EXISTS (SELECT 1 FROM issue_thread_interactions other_wait + WHERE other_wait.issue_id = ${issue}.id AND other_wait.status = 'pending' + AND (${passiveResponse('other_wait', issue)}) IS NOT TRUE)`; +} + +export function connectionIntentWaitingForResponse(interaction: string, issue: string): string { + return `${interaction}.company_id = ${issue}.company_id AND ${interaction}.issue_id = ${issue}.id + AND ${interaction}.kind = 'connection_intent' AND ${interaction}.status = 'pending' + AND ${issue}.status IN ('in_progress', 'in_review') AND ${issue}.assignee_user_id IS NULL + AND (${responseDrivenIssueWait(issue)})`; +} + +export function passiveBoardRecovery(alias: string): string { + return `${alias}.status IN ('active', 'escalated') AND ${alias}.resolved_at IS NULL + AND ${alias}.owner_type = 'board' AND ${alias}.owner_agent_id IS NULL + AND ${alias}.kind IN ('stranded_assigned_issue', 'missing_disposition') + AND ${alias}.cause IN ('stranded_assigned_issue', 'successful_run_missing_state', 'issue_review_path_lost') + AND ${alias}.recovery_issue_id IS NULL AND ${alias}.monitor_policy IS NULL AND ${alias}.timeout_at IS NULL + AND jsonb_typeof(${alias}.wake_policy) = 'object' + AND ${alias}.wake_policy->>'type' = 'board_escalation' + AND (${alias}.wake_policy - ARRAY['type', 'reason', 'preservesSourceAssignee']) = '{}'::jsonb + AND COALESCE(${alias}.wake_policy->>'preservesSourceAssignee', 'true') = 'true' + AND EXISTS (SELECT 1 FROM issues recovery_source + WHERE recovery_source.id = ${alias}.source_issue_id AND recovery_source.company_id = ${alias}.company_id + AND recovery_source.status = 'blocked' + AND recovery_source.checkout_run_id IS NULL AND recovery_source.execution_run_id IS NULL + AND recovery_source.conversation_agent_id IS NULL + AND recovery_source.monitor_next_check_at IS NULL AND recovery_source.monitor_wake_requested_at IS NULL + AND (recovery_source.execution_state IS NULL OR recovery_source.execution_state->>'status' = 'idle') + AND COALESCE(recovery_source.execution_policy->'monitor', 'null'::jsonb) = 'null'::jsonb)`; +} diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 5569f80c651..dd7173d9dd1 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -716,6 +716,7 @@ export interface PluginWorkerHandle { params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, + onWorkerSettled?: () => void, ): Promise; /** @@ -900,6 +901,7 @@ export function createPluginWorkerHandle( // Unlike RPC promises, these survive timeouts. A late response is the // completion receipt. A worker crash cannot manufacture one. const unsettledCalls = new Map void>(); + const unsettledInspections = new Set<() => void>(); let activeHostHandlers = 0; let idleHold: { ownerId: string; expiresAt: number; ready: boolean } | null = null; @@ -3018,6 +3020,10 @@ export function createPluginWorkerHandle( )), ); + // Read-only requests cannot continue in an exited worker. Do not infer + // completion for execute/cleanup calls, whose provider work may outlive it. + for (const finish of [...unsettledInspections]) finish(); + // Close the one login pseudo-terminal route with a fixed non-secret exit and // clear the route one time. The pending pseudo-terminal calls // already rejected through `rejectAllPending`. @@ -3351,14 +3357,17 @@ export function createPluginWorkerHandle( params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, + onWorkerSettled?: () => void, ): Promise { const rpcPromise = new Promise((resolve, reject) => { const idleControl = method === "prepareIdleSleep" || method === "releaseIdleSleep"; if (!idleControl && method !== "shutdown" && idleSleepHeld()) { + onWorkerSettled?.(); reject(new Error("Plugin is held for idle sleep")); return; } if (!childProcess?.stdin?.writable) { + onWorkerSettled?.(); reject( new Error( `Cannot call "${method}" — worker for "${pluginId}" is not running`, @@ -3368,7 +3377,20 @@ export function createPluginWorkerHandle( } const id = nextRequestId++; - if (!idleControl) unsettledCalls.set(id, beginIdleTrackedWork()); + if (!idleControl) { + const finishIdle = beginIdleTrackedWork(); + let completed = false; + const finish = () => { + if (completed) return; + completed = true; + finishIdle(); + unsettledInspections.delete(finish); + unsettledCalls.delete(id); + onWorkerSettled?.(); + }; + unsettledCalls.set(id, finish); + if (method === "environmentInspectLease") unsettledInspections.add(finish); + } const timeout = resolveRpcCallTimeoutMs(timeoutMs, rpcTimeoutMs); const invocationScope = deriveInvocationScope(method, params); const invocation = invocationScope ? registerInvocation(invocationScope) : null; @@ -3524,15 +3546,17 @@ export function createPluginWorkerHandle( params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, + onWorkerSettled?: () => void, ): Promise { if (status !== "running" && status !== "starting") { + onWorkerSettled?.(); return Promise.reject( new Error( `Cannot call "${method}" — worker for "${pluginId}" is ${status}`, ), ); } - return callInternal(method, params, timeoutMs, executeLogSink); + return callInternal(method, params, timeoutMs, executeLogSink, onWorkerSettled); }, openLoginPtySession(input: LoginPtyOpenInput) { @@ -3729,6 +3753,7 @@ export function createPluginWorkerManager( const log = logger.child({ service: "plugin-worker-manager" }); const workers = new Map(); const configApplied = new Set<() => void>(); + let activeInspections = 0; /** Per-plugin startup locks to prevent concurrent spawn races. */ const startupLocks = new Map>(); // The one shared, process-scoped aggregate route-slot controller. The manager @@ -3879,7 +3904,16 @@ export function createPluginWorkerManager( new Error(`No worker registered for plugin "${pluginId}"`), ); } - const result = handle.call(method, params, timeoutMs, executeLogSink); + let finishInspection: (() => void) | undefined; + if (method === "environmentInspectLease") { + if (activeInspections >= 2) return Promise.reject(new Error("Provider inspection concurrency limit reached")); + activeInspections += 1; + let finished = false; + finishInspection = () => { if (!finished) { finished = true; activeInspections -= 1; } }; + } + // A caller timeout is not worker completion. Inspection capacity is + // released only by its actual reply, send failure, or worker exit. + const result = handle.call(method, params, timeoutMs, executeLogSink, finishInspection); if (method === "configChanged") void result.then(() => { for (const listener of configApplied) { try { listener(); } diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index d92bae9560a..825e79e6bfe 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -2065,6 +2065,95 @@ export function routineService( return run; } + let statusReconcileCursor: string | null = null; + + async function syncRunStatusForIssue(issueId: string, executor: Db = db) { + const issue = await executor + .select({ + id: issues.id, + companyId: issues.companyId, + status: issues.status, + originKind: issues.originKind, + originRunId: issues.originRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + if (!issue || issue.originKind !== "routine_execution" || !issue.originRunId) return null; + const run = await executor + .select({ + id: routineRuns.id, + status: routineRuns.status, + failureReason: routineRuns.failureReason, + triggerPayload: routineRuns.triggerPayload, + }) + .from(routineRuns) + .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, issue.companyId))) + .where(and(eq(routineRuns.id, issue.originRunId), eq(routineRuns.companyId, issue.companyId))) + .then((rows) => rows[0] ?? null); + if (!run) return null; + if (issue.status === "done") { + const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) + ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); + const transientFailureClearedAt = executionIssueTransientFailureClearedAtFromPayload(run.triggerPayload); + return finalizeRun(issue.originRunId, { + status: "completed", + failureReason: null, + completedAt: new Date(), + ...(transientFailureStatus + ? { + triggerPayload: { + ...(run.triggerPayload ?? {}), + transientFailure: { + code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, + status: transientFailureStatus, + reason: executionIssueTransientFailureReason(transientFailureStatus), + clearedAt: transientFailureClearedAt ?? new Date().toISOString(), + }, + }, + } + : {}), + }, executor); + } + if (issue.status === "blocked" || issue.status === "cancelled") { + const failureReason = executionIssueTransientFailureReason(issue.status); + return finalizeRun(issue.originRunId, { + status: "failed", + failureReason, + completedAt: new Date(), + triggerPayload: { + ...(run.triggerPayload ?? {}), + transientFailure: { + code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, + status: issue.status, + reason: failureReason, + recordedAt: new Date().toISOString(), + }, + }, + }, executor); + } + const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) + ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); + if (run.status === "failed" && transientFailureStatus) { + return finalizeRun(issue.originRunId, { + status: "issue_created", + failureReason: null, + completedAt: null, + triggerPayload: { + ...(run.triggerPayload ?? {}), + transientFailure: { + code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, + status: transientFailureStatus, + reason: executionIssueTransientFailureReason(transientFailureStatus), + clearedAt: new Date().toISOString(), + }, + }, + }, executor); + } + return null; + } + + return { evaluateActivityGate, get: getRoutineById, @@ -3317,88 +3406,56 @@ export function routineService( return { triggered }; }, - syncRunStatusForIssue: async (issueId: string) => { - const issue = await db - .select({ - id: issues.id, - status: issues.status, - originKind: issues.originKind, - originRunId: issues.originRunId, - }) - .from(issues) - .where(eq(issues.id, issueId)) - .then((rows) => rows[0] ?? null); - if (!issue || issue.originKind !== "routine_execution" || !issue.originRunId) return null; - const run = await db - .select({ - id: routineRuns.id, - status: routineRuns.status, - failureReason: routineRuns.failureReason, - triggerPayload: routineRuns.triggerPayload, - }) - .from(routineRuns) - .where(eq(routineRuns.id, issue.originRunId)) - .then((rows) => rows[0] ?? null); - if (!run) return null; - if (issue.status === "done") { - const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) - ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); - const transientFailureClearedAt = executionIssueTransientFailureClearedAtFromPayload(run.triggerPayload); - return finalizeRun(issue.originRunId, { - status: "completed", - failureReason: null, - completedAt: new Date(), - ...(transientFailureStatus - ? { - triggerPayload: { - ...(run.triggerPayload ?? {}), - transientFailure: { - code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, - status: transientFailureStatus, - reason: executionIssueTransientFailureReason(transientFailureStatus), - clearedAt: transientFailureClearedAt ?? new Date().toISOString(), - }, - }, - } - : {}), - }); + // Repair a missed issue-status hook through the same status transition as + // normal issue writes. Never discard queued input or infer run completion + // from age. Lock both owners and re-read the issue before applying repair. + reconcileExecutionRunStatuses: async (): Promise<{ scanned: number; settled: number; deferred: number }> => { + const candidates = await db.select({ id: routineRuns.id, companyId: routineRuns.companyId, + issueId: issues.id, routineId: routineRuns.routineId }) + .from(routineRuns).innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, routineRuns.companyId))) + .innerJoin(issues, and(eq(issues.id, routineRuns.linkedIssueId), + eq(issues.companyId, routineRuns.companyId), eq(issues.originKind, "routine_execution"), + sql`${issues.originRunId} = ${routineRuns.id}::text`, sql`${issues.originId} = ${routineRuns.routineId}::text`)) + .where(and(eq(routineRuns.status, "issue_created"), inArray(issues.status, ["done", "cancelled", "blocked"]), + ...(statusReconcileCursor ? [sql`${routineRuns.id}::text > ${statusReconcileCursor}`] : []))) + .orderBy(asc(routineRuns.id)).limit(50); + statusReconcileCursor = candidates.length === 50 ? candidates.at(-1)!.id : null; + let settled = 0; + let deferred = 0; + for (const candidate of candidates) { + try { + const repaired = await db.transaction(async tx => { + await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`); + await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + const [routine] = await tx.select({ id: routines.id }).from(routines).where(and( + eq(routines.id, candidate.routineId), eq(routines.companyId, candidate.companyId))).for("share"); + if (!routine) return false; + const [issue] = await tx.select().from(issues).where(and(eq(issues.id, candidate.issueId), + eq(issues.companyId, candidate.companyId), eq(issues.originKind, "routine_execution"), + eq(issues.originRunId, candidate.id), eq(issues.originId, candidate.routineId))).for("update"); + if (!issue || !["done", "cancelled", "blocked"].includes(issue.status)) return false; + const [run] = await tx.select().from(routineRuns).where(and(eq(routineRuns.id, candidate.id), + eq(routineRuns.companyId, candidate.companyId), eq(routineRuns.routineId, candidate.routineId), + eq(routineRuns.linkedIssueId, issue.id), eq(routineRuns.status, "issue_created"))).for("update"); + if (!run) return false; + return (await syncRunStatusForIssue(issue.id, tx as unknown as Db)) !== null; + }); + if (repaired) settled += 1; + } catch (error) { + let current: unknown = error; + let retryable = false; + for (let depth = 0; depth < 5 && current && typeof current === "object"; depth++) { + const cause = current as { code?: string; cause?: unknown }; + if (["55P03", "57014", "40P01", "40001"].includes(cause.code ?? "")) { retryable = true; break; } + current = cause.cause; + } + if (!retryable) throw error; + deferred += 1; + } } - if (issue.status === "blocked" || issue.status === "cancelled") { - const failureReason = executionIssueTransientFailureReason(issue.status); - return finalizeRun(issue.originRunId, { - status: "failed", - failureReason, - completedAt: new Date(), - triggerPayload: { - ...(run.triggerPayload ?? {}), - transientFailure: { - code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, - status: issue.status, - reason: failureReason, - recordedAt: new Date().toISOString(), - }, - }, - }); - } - const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) - ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); - if (run.status === "failed" && transientFailureStatus) { - return finalizeRun(issue.originRunId, { - status: "issue_created", - failureReason: null, - completedAt: null, - triggerPayload: { - ...(run.triggerPayload ?? {}), - transientFailure: { - code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, - status: transientFailureStatus, - reason: executionIssueTransientFailureReason(transientFailureStatus), - clearedAt: new Date().toISOString(), - }, - }, - }); - } - return null; + return { scanned: candidates.length, settled, deferred }; }, + + syncRunStatusForIssue, }; }