From b762036c984be27ba8aa0c4395e6395741d1e848 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:01:23 -0500 Subject: [PATCH 1/6] fix(server): schedule native backup expiration from retention deadlines Co-Authored-By: Paperclip --- .../2026-10-30-backup-expiration-sleep.md | 44 +++++ .../native-harness-backup-expiration.test.ts | 150 ++++++++++++++++++ server/src/index.ts | 36 +---- .../services/delivery-work-notifications.ts | 1 + server/src/services/issues.ts | 4 + ...ve-environment-settlement-notifications.ts | 2 + ...harness-backup-expiration-notifications.ts | 19 +++ .../native-harness-backup-expiration-work.ts | 53 +++++++ .../native-harness-backup-expiration.ts | 130 ++++++++++----- .../native-runtime/native-session-executor.ts | 3 + ...ed-native-session-cleanup-notifications.ts | 2 + 11 files changed, 372 insertions(+), 72 deletions(-) create mode 100644 doc/plans/2026-10-30-backup-expiration-sleep.md create mode 100644 server/src/services/native-harness-backup-expiration-notifications.ts create mode 100644 server/src/services/native-runtime/native-harness-backup-expiration-work.ts diff --git a/doc/plans/2026-10-30-backup-expiration-sleep.md b/doc/plans/2026-10-30-backup-expiration-sleep.md new file mode 100644 index 00000000000..47d41bbf581 --- /dev/null +++ b/doc/plans/2026-10-30-backup-expiration-sleep.md @@ -0,0 +1,44 @@ +# Backup expiration without idle polling + +The backup expiration duty retains its existing minimum age and safety policy. +It uses the shared scoped deadline coordinator. The archived runner evidence and +checkpoint expiration duties keep their separate registrations. + +The durable source is each controller-owned `failover-backups/current` directory, +its control-plane session identity, and the associated run, issue, finalization, +and environment lease rows. The next eligible instant is one millisecond after +the later of the backup mtime and every related issue's close time, plus the +configured retention duration. An open issue, live run or lease, unsettled +finalization, or cleanup activation has no time-derived expiry; its producer +must invalidate the source when it changes. A busy process or failed I/O leaves +real unresolved cleanup and uses the shared retry backoff. Dry-run reports an +eligible source once and waits for source invalidation or restart. + +Startup pages existing disk scopes. Roots without backup or cleanup tombstones +perform no backup SQL. After discovery and all obligations settle, there is no +recurring SQL or filesystem scan. Startup reconstructs future work; inspecting a +source never changes its business timestamps. Runtime task close/reopen and +retention timestamp changes, native environment and retained session writes, +local backup publication, and process release invalidate the company or digest. +Signals from database producers register before writes inside their owning +transaction, including caller-owned transactions and nested savepoints. + +The controller remains the only supported writer of its local backup files. +An independent process writing those files must use the same producer ingress; +manual disk edits require reconstruction on restart. Database writer ingress +continues to use the existing transaction-aware work signals. This change adds +no permanent database listener and does not add a polling table or migration. + +The existing issue share locks now remain held through the retirement rename. +This fences task reopen while the backup becomes unavailable. A failed database +transaction restores `.retiring-*`; a process interruption is recovered from +that existing file journal at startup. After database settlement, `.expired-*` +records outstanding physical removal so a partial delete retries idempotently. +Ownership, session identity, live run, lease and local execution safety checks +remain in the backup owner. No host-sleep blocker or expiry-on-use rule changes. +An in-process deadline does not register a Cloud wake alarm. + +Acceptance uses isolated real PostgreSQL for service writes, transaction signals, +safety checks, resume pointer deletion, and restart reconstruction. A temporary +filesystem is the controlled storage boundary. These tests do not claim cloud +object storage behavior, deployed sleep, or fleet savings. diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index 2437d9711b5..be38a9ae0d9 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -1,3 +1,7 @@ +import { registerNativeHarnessBackupExpirationWork } from "../services/native-runtime/native-harness-backup-expiration-work.js"; +import { noteNativeHarnessBackupChange, notifyNativeHarnessBackupExpiration } from "../services/native-harness-backup-expiration-notifications.js"; +import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; +import { issueService } from "../services/issues.js"; import { applyPendingMigrations } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { registerNativeRunnerEvidenceRetentionWork } from "../services/native-runtime/native-runner-evidence-retention-work.js"; @@ -55,7 +59,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => { let coordinator: ReturnType | undefined; let worker: ReturnType | undefined; let admitted = true; + let backupWorker: ReturnType | undefined; + const backupErrors: unknown[] = []; afterEach(async () => { + await backupWorker?.stop(); backupWorker = undefined; backupErrors.length = 0; worker?.unsubscribe(); await coordinator?.stop(); coordinator = undefined; worker = undefined; admitted = true; vi.useRealTimers(); vi.restoreAllMocks(); await db.delete(agentTaskSessions); @@ -558,6 +565,149 @@ describeEmbeddedPostgres("native harness backup expiration", () => { reads.mockRestore(); writes.mockRestore(); directories.mockRestore(); } + async function startBackups(options: Parameters[1] = {}) { + if (!stateBase) await emptyStateBase(); + if (!vi.isFakeTimers()) { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + vi.setSystemTime(now); + } + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => true, + onError: error => backupErrors.push(error) }); + backupWorker = registerNativeHarnessBackupExpirationWork(coordinator, db, + { stateBase, isSessionBusy: () => false, ...options }, () => admitted); + await backupWorker.ready; + } + + it.each([false, true])("backup deadline worker leaves empty and retained terminal history quiet (%s)", async history => { + await emptyStateBase(); + if (history) { + const session = await seedSession(); + await fs.rm(path.join(session.root, "failover-backups"), { recursive: true }); + for (let index = 0; index < 250; index++) await fs.mkdir(path.join(stateBase, + createHash("sha256").update(String(index)).digest("hex"))); + } + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await startBackups({ maxEntries: 20 }); await settled(); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); await quiet(); + expect(backupErrors).toEqual([]); + }); + + it("ordinary task close arms exact backup retention, deletes physical bytes and resume pointer, then stays quiet", async () => { + const session = await seedSession({ issueStatus: "todo" }); + await startBackups(); await settled(); await quiet(); + const closed = await issueService(db).update(session.issueId, { status: "done", companyGuard: session.companyId }); + const deadline = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1; + await settled(deadline); await quiet(60_000); + await vi.advanceTimersByTimeAsync(deadline - Date.now() - 1); + expect(await exists(session.current)).toBe(true); + expect(await taskSessionCount(session.agentId)).toBe(1); + await vi.advanceTimersByTimeAsync(1); await settled(); + expect(await exists(session.current)).toBe(false); + expect(await taskSessionCount(session.agentId)).toBe(0); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, session.runId))).toHaveLength(1); + expect(backupErrors).toEqual([]); await quiet(); + }); + + it("reopen cancels a saved deadline and a later ordinary close restores retention", async () => { + const session = await seedSession({ closedDaysAgo: 2 }); + await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1); + await issueService(db).update(session.issueId, { status: "todo" }); + await settled(); await quiet(8 * DAY_MS); expect(await exists(session.current)).toBe(true); + const closed = await issueService(db).update(session.issueId, { status: "done" }); + const due = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1; + await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); await quiet(); + }); + + it("reconstructs a retained backup deadline on restart without changing business timestamps", async () => { + const session = await seedSession({ closedDaysAgo: 2 }); + const [before] = await db.select().from(issues).where(eq(issues.id, session.issueId)); + const due = now.getTime() + 5 * DAY_MS + 1; + await startBackups(); await settled(due); + await backupWorker!.stop(); await coordinator!.stop(); + await startBackups(); await settled(due); + const [after] = await db.select().from(issues).where(eq(issues.id, session.issueId)); + expect(after!.updatedAt).toEqual(before!.updatedAt); expect(after!.completedAt).toEqual(before!.completedAt); + await quiet(); + }); + + it("reports dry-run backups once and rechecks only after source invalidation", async () => { + const session = await seedSession(); const report = vi.spyOn(logger, "info"); + await startBackups({ dryRun: true }); await settled(); + expect(report.mock.calls.filter(call => call[1] === "native harness backup expiration")).toHaveLength(1); + expect(await taskSessionCount(session.agentId)).toBe(1); expect(await exists(session.current)).toBe(true); + await quiet(); + noteNativeHarnessBackupChange(db, { companyId: session.companyId, scopeDigest: session.scope }); + await settled(); + expect(report.mock.calls.filter(call => call[1] === "native harness backup expiration")).toHaveLength(2); + await quiet(); + }); + + it("retains live lease references and only removes the owning company's backup after lease release", async () => { + const first = await seedSession({ leaseStatus: "retained" }); + const other = await seedSession({ leaseStatus: "retained" }); + await startBackups(); await settled(); await quiet(); + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "expired" }) + .where(eq(environmentLeases.companyId, first.companyId)), { companyId: first.companyId }); + await settled(); expect(await exists(first.current)).toBe(false); expect(await exists(other.current)).toBe(true); + expect(await taskSessionCount(other.agentId)).toBe(1); await quiet(); + }); + + it("retries a partial physical deletion from its tombstone without recreating a resume pointer", async () => { + const session = await seedSession(); const remove = fs.rm; let failed = false; + const fault = vi.spyOn(fs, "rm").mockImplementation(async (...args) => { + if (!failed && String(args[0]).includes(".expired-")) { failed = true; throw new Error("controlled filesystem delete failure"); } + return remove(...args); + }); + await startBackups(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(100); expect(backupErrors).toHaveLength(1); }); + const retry = coordinator!.nextWakeAt()!; expect(retry - Date.now()).toBeGreaterThan(20_000); + expect(await taskSessionCount(session.agentId)).toBe(0); expect(await exists(session.current)).toBe(false); + fault.mockRestore(); await quiet(5000); + await vi.advanceTimersByTimeAsync(retry - Date.now()); await settled(); + expect(await fs.readdir(path.join(session.root, "failover-backups"))).toEqual([]); await quiet(); + }); + + it("retains a first ordinary close during drain and dispatches it after resume", async () => { + const session = await seedSession({ issueStatus: "todo" }); + await startBackups(); await settled(); admitted = false; + const closed = await issueService(db).update(session.issueId, { status: "done" }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + await quiet(60_000); admitted = true; + const due = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1; + await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); await quiet(); + }); + + it("honors outer rollback for nested ordinary close and activates only on committed work", async () => { + const session = await seedSession({ issueStatus: "todo" }); + await startBackups(); await settled(); + await expect(db.transaction(async outer => { + await outer.transaction(async nested => { + await issueService(db).update(session.issueId, { status: "done" }, nested); + }); + expect(await exists(session.current)).toBe(true); + throw new Error("rollback outer close"); + })).rejects.toThrow("rollback outer close"); + await settled(); await quiet(8 * DAY_MS); + expect(await exists(session.current)).toBe(true); + const [saved] = await db.select().from(issues).where(eq(issues.id, session.issueId)); + expect(saved!.status).toBe("todo"); + }); + + it("pages multiple company backups, preserves a busy one and wakes from process release", async () => { + const sessions = [await seedSession(), await seedSession(), await seedSession()]; let busy = true; + await startBackups({ maxEntries: 1, maxRemovals: 1, isSessionBusy: digest => busy && digest === sessions[1]!.scope }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); + expect(await exists(sessions[0]!.current)).toBe(false); expect(await exists(sessions[2]!.current)).toBe(false); + expect(backupErrors.length).toBeGreaterThan(0); + }); + expect(await exists(sessions[1]!.current)).toBe(true); busy = false; + noteNativeHarnessBackupChange(db, { companyId: sessions[1]!.companyId }); + await settled(); expect(await exists(sessions[1]!.current)).toBe(false); await quiet(); + }); + it("leaves empty evidence and retained backups quiet without invoking backup SQL", async () => { const session = await seedSession(); const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); diff --git a/server/src/index.ts b/server/src/index.ts index ae229f03518..919af75f14a 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerNativeHarnessBackupExpirationWork } from "./services/native-runtime/native-harness-backup-expiration-work.js"; import { registerProviderTraceRetentionWork } from "./services/provider-trace-retention-work.js"; import { registerRunEvidenceMigrationWork } from "./services/run-evidence-migration-work.js"; import { registerSkillVersionRetentionWork } from "./services/skill-version-retention-work.js"; @@ -32,8 +33,6 @@ import { skillVersionRetentionService, } from "./services/skill-version-retention.js"; import { - NATIVE_HARNESS_BACKUP_EXPIRATION_INTERVAL_MS, - NATIVE_HARNESS_BACKUP_EXPIRATION_STARTUP_DELAY_MS, nativeHarnessBackupExpirationService, } from "./services/native-runtime/native-harness-backup-expiration.js"; import { registerNativeRunnerEvidenceRetentionWork } from "./services/native-runtime/native-runner-evidence-retention-work.js"; @@ -1290,30 +1289,6 @@ async function startServerWithDatabaseTeardown( evidenceRetentionDays: config.nativeRunnerEvidenceRetentionDays, shouldContinue: () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby(), }); - let nativeHarnessBackupExpirationInFlight = false; - let nativeHarnessBackupExpirationContinuation: ReturnType | undefined; - const sweepNativeHarnessBackups = () => { - if (!config.nativeHarnessBackupExpirationEnabled || nativeHarnessBackupExpirationInFlight || isIdleTaskDrainActive() || heartbeatSchedulerStopped || isWarmStandby()) return; - nativeHarnessBackupExpirationInFlight = true; - trackHeartbeatSchedulerWork(nativeHarnessBackupExpiration.tick() - .then((result) => { - if (result.candidates > 0 || result.evidenceCandidates > 0 || result.failed > 0) { - logger.info({ ...result, dryRun: config.nativeHarnessBackupExpirationDryRun }, "native harness backup expiration"); - } - if (result.truncated && !heartbeatSchedulerStopped) { - nativeHarnessBackupExpirationContinuation = setTimeout(sweepNativeHarnessBackups, 1000); - nativeHarnessBackupExpirationContinuation.unref?.(); - } - }) - .catch((err) => { - logger.error({ err }, "native harness backup expiration failed"); - }) - .finally(() => { nativeHarnessBackupExpirationInFlight = false; })); - }; - const nativeHarnessBackupExpirationInterval = setInterval(sweepNativeHarnessBackups, NATIVE_HARNESS_BACKUP_EXPIRATION_INTERVAL_MS); - nativeHarnessBackupExpirationInterval.unref?.(); - const nativeHarnessBackupExpirationStartup = setTimeout(sweepNativeHarnessBackups, NATIVE_HARNESS_BACKUP_EXPIRATION_STARTUP_DELAY_MS); - nativeHarnessBackupExpirationStartup.unref?.(); const externalObjects = externalObjectService(db as any, { pluginWorkerManager, enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true, @@ -1359,6 +1334,12 @@ async function startServerWithDatabaseTeardown( const deliveryWork = app.locals.deliveryWork as ReturnType; if (config.nativeHarnessBackupExpirationEnabled) { nativeHarnessBackupExpiration.registerCheckpointExpiration(deliveryWork); + const backupExpirationWork = registerNativeHarnessBackupExpirationWork(deliveryWork, db as any, { + dryRun: config.nativeHarnessBackupExpirationDryRun, + closedForDays: config.nativeHarnessBackupExpirationDays, + }, () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby()); + trackHeartbeatSchedulerWork(backupExpirationWork.ready); + server.on("close", backupExpirationWork.stop); } const taskDrainExpiryWork = registerTaskDrainExpiryWork(deliveryWork, () => { if (heartbeatSchedulerStopped || isWarmStandby()) return; @@ -2235,9 +2216,6 @@ async function startServerWithDatabaseTeardown( heartbeatSchedulerStopped = true; skillVersionRetentionWorker?.unsubscribe(); await taskDrainExpiryWork.stop(); - clearInterval(nativeHarnessBackupExpirationInterval); - clearTimeout(nativeHarnessBackupExpirationStartup); - clearTimeout(nativeHarnessBackupExpirationContinuation); nativeRunnerEvidenceRetentionWorker?.unsubscribe(); await skillSourceSync.stop(); unsubscribeChatCompletions(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9195f1fcb12..4a4d66edee1 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -118,6 +118,7 @@ export const DELIVERY_QUEUES = { issueMonitors: "issue-monitors", agentTimers: "agent-timers", caseHandoffs: "case-handoffs", + nativeHarnessBackupExpiration: "native-harness-backup-expiration", nativeCheckpointExpiration: "native-checkpoint-expiration", } as const; export type DeliveryQueue = typeof DELIVERY_QUEUES[keyof typeof DELIVERY_QUEUES]; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 42c65bb5d02..fdb02c5ea9e 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyNativeHarnessBackupExpiration } from "./native-harness-backup-expiration-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -11486,6 +11487,9 @@ export function issueService(db: Db) { (blockedByIssueIds !== undefined || issueData.unblockDescriptor !== undefined)))) { patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number; } + if (patch.status !== undefined || patch.completedAt !== undefined || patch.cancelledAt !== undefined) { + await notifyNativeHarnessBackupExpiration(tx, { companyId: receiptExisting.companyId }); + } if ((patch.status !== undefined && patch.status !== receiptExisting.status) || patch.completedAt !== undefined || patch.cancelledAt !== undefined || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) { diff --git a/server/src/services/native-environment-settlement-notifications.ts b/server/src/services/native-environment-settlement-notifications.ts index 5769b81015a..427c43522b7 100644 --- a/server/src/services/native-environment-settlement-notifications.ts +++ b/server/src/services/native-environment-settlement-notifications.ts @@ -1,8 +1,10 @@ +import { notifyNativeHarnessBackupExpiration } from "./native-harness-backup-expiration-notifications.js"; import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string }; export async function notifyNativeEnvironmentSettlement(tx: object, scope?: NativeEnvironmentSettlementScope) { + await notifyNativeHarnessBackupExpiration(tx, scope && { companyId: scope.companyId }); if (scope) await notifySkillVersionRetention(tx, scope.companyId); await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeEnvironmentSettlement, scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"); diff --git a/server/src/services/native-harness-backup-expiration-notifications.ts b/server/src/services/native-harness-backup-expiration-notifications.ts new file mode 100644 index 00000000000..602a7063acd --- /dev/null +++ b/server/src/services/native-harness-backup-expiration-notifications.ts @@ -0,0 +1,19 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeHarnessBackupExpirationScope = { companyId: string; scopeDigest?: string }; + +export function notifyNativeHarnessBackupExpiration(tx: object, scope?: NativeHarnessBackupExpirationScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeHarnessBackupExpiration, scope ? JSON.stringify(scope) : "*"); +} + +const listeners = new WeakMap void>>(); +export function onNativeHarnessBackupChange(db: object, listener: (scope: NativeHarnessBackupExpirationScope) => void) { + let owned = listeners.get(db); + if (!owned) { owned = new Set(); listeners.set(db, owned); } + owned.add(listener); + return () => { owned.delete(listener); if (!owned.size) listeners.delete(db); }; +} + +export function noteNativeHarnessBackupChange(db: object, scope: NativeHarnessBackupExpirationScope) { + for (const listener of listeners.get(db) ?? []) listener(scope); +} diff --git a/server/src/services/native-runtime/native-harness-backup-expiration-work.ts b/server/src/services/native-runtime/native-harness-backup-expiration-work.ts new file mode 100644 index 00000000000..93c4f08e340 --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-expiration-work.ts @@ -0,0 +1,53 @@ +import type { Db } from "@paperclipai/db"; +import { logger } from "../../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { onNativeHarnessBackupChange, type NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js"; +import { onRetainedNativeSessionCleanupRelease } from "../retained-native-session-cleanup-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { nativeHarnessBackupExpirationService, type NativeHarnessBackupExpirationOptions } from "./native-harness-backup-expiration.js"; + +type Entry = { id: string; companyId: string; agentId: string; dueAt: number }; + +export function registerNativeHarnessBackupExpirationWork( + coordinator: ReturnType, db: Db, + options: NativeHarnessBackupExpirationOptions, canRun: () => boolean, +) { + const service = nativeHarnessBackupExpirationService(db, { ...options, duty: "backups" }); + const cursors = new Map(); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeHarnessBackupExpiration, batchSize: options.maxRemovals ?? 25, + decodeScope: (hint: string): NativeHarnessBackupExpirationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativeHarnessBackupExpirationScope) => !scope || entry.companyId === scope.companyId + && (!scope.scopeDigest || entry.id === scope.scopeDigest), + load: async (scope, signal) => { + const key = scope ? JSON.stringify(scope) : "*"; + const page = scope?.scopeDigest ? { names: [scope.scopeDigest], complete: true } + : await service.backupScopes(cursors.get(key), options.maxEntries ?? 100); + const rows: Entry[] = []; + for (const name of page.names) { + if (signal?.aborted || !canRun()) throw new Error("Backup expiration discovery paused"); + const entry = await service.inspectBackup(name); + if (entry && (!scope || entry.companyId === scope.companyId)) rows.push(entry); + } + if (page.complete) cursors.delete(key); else cursors.set(key, page.names.at(-1)!); + return { rows, complete: page.complete }; + }, + check: async (entry, signal) => { + if (signal?.aborted || !canRun()) throw new Error("Backup expiration paused"); + const current = await service.inspectBackup(entry.id); + if (!current || current.companyId !== entry.companyId) return null; + if (current.dueAt > Date.now()) return current; + const result = await service.expireBackup(entry.id); + if (result.candidates || result.failed) logger.info({ ...result, dryRun: options.dryRun ?? false }, "native harness backup expiration"); + if (options.dryRun) return null; + const remaining = await service.inspectBackup(entry.id); + if (!remaining) return null; + if (result.skipped.session_active) throw new Error("Backup session is active"); + return remaining; + }, + }, canRun); + const unsubscribe = onNativeHarnessBackupChange(db, scope => worker.invalidate(JSON.stringify(scope))); + const release = onRetainedNativeSessionCleanupRelease(db, scope => worker.invalidate(JSON.stringify({ companyId: scope.companyId }))); + return { ...worker, async stop() { unsubscribe(); release(); await worker.stop(); } }; +} diff --git a/server/src/services/native-runtime/native-harness-backup-expiration.ts b/server/src/services/native-runtime/native-harness-backup-expiration.ts index 8ea9a242629..2d45b12135f 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration.ts @@ -170,7 +170,7 @@ type Tx = Parameters[0]>[0]; async function sessionRetentionBlocker( tx: Tx, input: { scopeDigest: string | null; identity: SessionIdentity; closedBefore: Date; checkpoint?: boolean }, -): Promise<{ reason: string } | { companyId: string; agentId: string; runIds: string[] }> { +): Promise<{ reason: string } | { companyId: string; agentId: string; runIds: string[]; issueIds: string[]; closedAt: number }> { const [owner] = await tx .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) .from(heartbeatRuns) @@ -246,7 +246,7 @@ async function sessionRetentionBlocker( .where(and(inArray(environmentLeases.status, LIVE_LEASE_STATUSES), or(...leaseReferences))) .limit(1); if (lease) return { reason: "live_lease" }; - return { ...owner, runIds }; + return { ...owner, runIds, issueIds, closedAt: Math.max(...issueRows.map(issue => new Date(issue.closedAt!).getTime())) }; } async function hasLiveRun(db: Db, owner: { companyId: string; agentId: string }, sessionId: string) { @@ -379,54 +379,48 @@ export function nativeHarnessBackupExpirationService( const identity = await readSessionIdentity(root); if (!identity) return skip("identity_unknown"); - const decision = await db.transaction(async (tx) => { - const blocker = await sessionRetentionBlocker(tx, { - scopeDigest, - identity, - closedBefore: options.closedBefore, - }); - if ("reason" in blocker) return blocker; - if (options.dryRun) return { owner: blocker, cleared: 0 }; - await notifyExternalConversationSettlement(tx, { companyId: blocker.companyId }); - const cleared = await tx - .delete(agentTaskSessions) - .where( - and( - eq(agentTaskSessions.companyId, blocker.companyId), - eq(agentTaskSessions.agentId, blocker.agentId), - sql`${agentTaskSessions.sessionParamsJson}->>'sessionId' = ${identity.normalizedSessionId}`, - ), - ) - .returning({ id: agentTaskSessions.id }); - return { owner: blocker, cleared: cleared.length }; - }); - if ("reason" in decision) return skip(decision.reason); - - const bytes = await allocatedBytes(current); - result.candidates++; - result.candidateBytes += bytes; - if (options.dryRun) return; - result.clearedTaskSessions += decision.cleared; - if (options.isSessionBusy(scopeDigest)) return skip("session_active"); const id = randomUUID(); const retiring = path.join(backupRoot, `${RETIRING_PREFIX}${id}`); const trash = path.join(backupRoot, `${EXPIRED_PREFIX}${id}`); - await fs.rename(current, retiring); - let liveRun: boolean; + let retired = false; + let decision: { reason: string } | { cleared: number; bytes: number }; try { - liveRun = await hasLiveRun(db, decision.owner, identity.normalizedSessionId); - } catch (err) { - await fs.rename(retiring, current); - throw err; - } - if (liveRun) { - await fs.rename(retiring, current); - return skip("live_run"); + decision = await db.transaction(async (tx) => { + const blocker = await sessionRetentionBlocker(tx, { + scopeDigest, identity, closedBefore: options.closedBefore, + }); + if ("reason" in blocker) return blocker; + const bytes = await allocatedBytes(current); + if (options.dryRun) return { cleared: 0, bytes }; + if (options.isSessionBusy(scopeDigest)) return { reason: "session_active" }; + await notifyExternalConversationSettlement(tx, { companyId: blocker.companyId }); + await fs.rename(current, retiring); + retired = true; + if (await hasLiveRun(tx as unknown as Db, blocker, identity.normalizedSessionId)) { + await fs.rename(retiring, current); + retired = false; + return { reason: "live_run" }; + } + const cleared = await tx.delete(agentTaskSessions).where(and( + eq(agentTaskSessions.companyId, blocker.companyId), + eq(agentTaskSessions.agentId, blocker.agentId), + sql`${agentTaskSessions.sessionParamsJson}->>'sessionId' = ${identity.normalizedSessionId}`, + )).returning({ id: agentTaskSessions.id }); + return { cleared: cleared.length, bytes }; + }); + } catch (error) { + if (retired) await fs.rename(retiring, current); + throw error; } + if ("reason" in decision) return skip(decision.reason); + result.candidates++; + result.candidateBytes += decision.bytes; + if (options.dryRun) return; + result.clearedTaskSessions += decision.cleared; await fs.rename(retiring, trash); await removeBackupTree(trash); result.expired++; - result.reclaimedBytes += bytes; + result.reclaimedBytes += decision.bytes; } async function removeExpiredLeftovers(parent: string, context: EvidenceContext, result: NativeHarnessBackupExpirationResult) { @@ -745,5 +739,55 @@ export function nativeHarnessBackupExpirationService( }); } - return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration }; + async function inspectBackup(scopeDigest: string) { + const base = path.resolve(defaults.stateBase ?? nativeHarnessStateBase()); + const root = path.join(base, scopeDigest); + if (!SCOPE_PATTERN.test(scopeDigest) || !(await isRealDirectory(root))) return null; + const backupRoot = path.join(root, "failover-backups"); + if (!(await isRealDirectory(backupRoot))) return null; + const names = await fs.readdir(backupRoot); + const current = path.join(backupRoot, "current"); + const stat = await fs.lstat(current).catch(missingPath); + const leftovers = names.some(name => [RETIRING_PREFIX, EXPIRED_PREFIX].some(prefix => + name.startsWith(prefix) && UUID_PATTERN.test(name.slice(prefix.length)))); + if ((!stat?.isDirectory() || stat.isSymbolicLink()) && !leftovers) return null; + if (defaults.dryRun && !stat?.isDirectory()) return null; + const identity = await readSessionIdentity(root); + if (!identity) return null; + const [owner] = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); + if (!owner) return null; + if (leftovers && !defaults.dryRun) return { id: scopeDigest, ...owner, dueAt: 0 }; + if (await fs.lstat(path.join(root, "cleanup-activation.json")).then(() => true, () => false)) return null; + const blocker = await db.transaction(tx => sessionRetentionBlocker(tx, { + scopeDigest, identity, closedBefore: new Date(8_000_000_000_000_000), + })); + if ("reason" in blocker) return null; + const retentionMs = Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS; + return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtimeMs, blocker.closedAt) + retentionMs + 1 }; + } + + async function backupScopes(after?: string, limit = 100) { + const base = path.resolve(defaults.stateBase ?? nativeHarnessStateBase()); + const names = (await fs.readdir(base).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return [] as string[]; + throw error; + })).filter(name => SCOPE_PATTERN.test(name) && (!after || name > after)).sort(); + return { names: names.slice(0, limit), complete: names.length <= limit }; + } + + async function expireBackup(scopeDigest: string) { + const result = emptyResult(); + const root = path.join(path.resolve(defaults.stateBase ?? nativeHarnessStateBase()), scopeDigest); + const dryRun = defaults.dryRun ?? false; + await expireSession(root, scopeDigest, { + dryRun, + isSessionBusy: defaults.isSessionBusy ?? nativeSessionScopeDigestIsBusy, + closedBefore: new Date(Date.now() - Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS), + }, result); + return result; + } + + return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration, + inspectBackup, backupScopes, expireBackup }; } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 4f5c5ae99de..b47ddc40fee 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { noteNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeTurnAccounting, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent } from "./native-usage-accounting.js"; export { nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage } from "./native-usage-accounting.js"; @@ -13310,6 +13311,8 @@ async function createRunnerdBackendWithinSessionClaim( throw error; } removeNativeHarnessBackup(previousRoot); + noteNativeHarnessBackupChange(input.db, { companyId: input.execution.binding.companyId, + scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex") }); } finally { removeNativeHarnessBackup(pendingRoot); } diff --git a/server/src/services/retained-native-session-cleanup-notifications.ts b/server/src/services/retained-native-session-cleanup-notifications.ts index 1ff39143e0e..d5aaff7c31a 100644 --- a/server/src/services/retained-native-session-cleanup-notifications.ts +++ b/server/src/services/retained-native-session-cleanup-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeHarnessBackupExpiration } from "./native-harness-backup-expiration-notifications.js"; import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -9,6 +10,7 @@ export type RetainedNativeSessionCleanupScope = { }; export async function notifyRetainedNativeSessionCleanup(tx: object, scope: RetainedNativeSessionCleanupScope) { + await notifyNativeHarnessBackupExpiration(tx, { companyId: scope.companyId }); await notifySkillVersionRetention(tx, scope.companyId); await notifyDeliveryWork(tx, DELIVERY_QUEUES.retainedNativeSessionCleanup, JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId, agentId: scope.agentId })); } From 79d1e3a404aa929cb5182d9e37cb23f2fa2ac52a Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:07:12 -0500 Subject: [PATCH 2/6] test(server): qualify native backup publication and retention recovery Co-Authored-By: Paperclip --- .../native-harness-backup-expiration.test.ts | 69 ++++++++++++++++++- .../native-harness-backup-expiration.ts | 2 +- .../native-harness-backup-publication.test.ts | 44 ++++++++++++ .../native-harness-backup-publication.ts | 28 ++++++++ .../native-runtime/native-session-executor.ts | 43 +++--------- 5 files changed, 149 insertions(+), 37 deletions(-) create mode 100644 server/src/services/native-runtime/native-harness-backup-publication.test.ts create mode 100644 server/src/services/native-runtime/native-harness-backup-publication.ts diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index be38a9ae0d9..1ac2539545e 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -1,5 +1,6 @@ +import { publishNativeHarnessBackup } from "../services/native-runtime/native-harness-backup-publication.js"; import { registerNativeHarnessBackupExpirationWork } from "../services/native-runtime/native-harness-backup-expiration-work.js"; -import { noteNativeHarnessBackupChange, notifyNativeHarnessBackupExpiration } from "../services/native-harness-backup-expiration-notifications.js"; +import { noteNativeHarnessBackupChange } from "../services/native-harness-backup-expiration-notifications.js"; import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; import { issueService } from "../services/issues.js"; import { applyPendingMigrations } from "@paperclipai/db"; @@ -583,6 +584,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => { if (history) { const session = await seedSession(); await fs.rm(path.join(session.root, "failover-backups"), { recursive: true }); + await db.insert(heartbeatRuns).values(Array.from({ length: 500 }, () => ({ + companyId: session.companyId, agentId: session.agentId, status: "succeeded", issueId: session.issueId, + nativeIssueId: session.issueId, nativeSessionId: randomUUID(), finishedAt: now, + }))); for (let index = 0; index < 250; index++) await fs.mkdir(path.join(stateBase, createHash("sha256").update(String(index)).digest("hex"))); } @@ -609,6 +614,68 @@ describeEmbeddedPostgres("native harness backup expiration", () => { expect(backupErrors).toEqual([]); await quiet(); }); + it("ordinary backup publication after quiet schedules its actual file age and leaves a durable cleanup result", async () => { + const session = await seedSession(); + const backupRoot = path.dirname(session.current); + const pendingRoot = path.join(backupRoot, "pending-publication"); + await fs.rename(session.current, pendingRoot); + await startBackups(); await settled(); await quiet(); + const publishedAt = Date.now(); + await fs.utimes(pendingRoot, new Date(publishedAt), new Date(publishedAt)); + await publishNativeHarnessBackup({ db, companyId: session.companyId, scopeDigest: session.scope, backupRoot, pendingRoot }); + const stat = await fs.stat(session.current); + const due = stat.mtime.getTime() + 7 * DAY_MS + 1; + await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); expect(await taskSessionCount(session.agentId)).toBe(0); + expect(await fs.readdir(backupRoot)).toEqual([]); await quiet(); + }); + + it("a committed earlier close timestamp replaces the later deadline without early cleanup", async () => { + const session = await seedSession({ closedDaysAgo: 2 }); + await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1); + const earlier = new Date(now.getTime() - 6 * DAY_MS); + await issueService(db).update(session.issueId, { completedAt: earlier }); + const due = earlier.getTime() + 7 * DAY_MS + 1; + await settled(due); await quiet(60_000); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); await quiet(); + }); + + it("holds nested ordinary close activation until the outer transaction commits", async () => { + const session = await seedSession({ issueStatus: "todo" }); + await startBackups(); await settled(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const staged = new Promise(resolve => { written = resolve; }); + const transaction = db.transaction(async outer => { + await outer.transaction(nested => issueService(db).update(session.issueId, { status: "done" }, nested)); + written(); await gate; + }); + await staged; + await vi.advanceTimersByTimeAsync(60_000); + expect(await exists(session.current)).toBe(true); + expect(await taskSessionCount(session.agentId)).toBe(1); + release(); await transaction; + const [closed] = await db.select().from(issues).where(eq(issues.id, session.issueId)); + const due = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1; + await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); await quiet(); + }); + + it("preserves failed cleanup retry across drain, then finishes its physical tombstone on resume", async () => { + const session = await seedSession(); const remove = fs.rm; + const fault = vi.spyOn(fs, "rm").mockImplementation(async (...args) => { + if (String(args[0]).includes(".expired-")) throw new Error("controlled drain retry failure"); + return remove(...args); + }); + await startBackups(); + expect(backupErrors).toHaveLength(1); admitted = false; fault.mockRestore(); + await quiet(60_000); + expect((await fs.readdir(path.dirname(session.current))).some(name => name.startsWith(".expired-"))).toBe(true); + admitted = true; await settled(); + expect(await fs.readdir(path.dirname(session.current))).toEqual([]); await quiet(); + }); + it("reopen cancels a saved deadline and a later ordinary close restores retention", async () => { const session = await seedSession({ closedDaysAgo: 2 }); await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1); diff --git a/server/src/services/native-runtime/native-harness-backup-expiration.ts b/server/src/services/native-runtime/native-harness-backup-expiration.ts index 2d45b12135f..09ddea9e5d2 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration.ts @@ -764,7 +764,7 @@ export function nativeHarnessBackupExpirationService( })); if ("reason" in blocker) return null; const retentionMs = Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS; - return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtimeMs, blocker.closedAt) + retentionMs + 1 }; + return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 }; } async function backupScopes(after?: string, limit = 100) { diff --git a/server/src/services/native-runtime/native-harness-backup-publication.test.ts b/server/src/services/native-runtime/native-harness-backup-publication.test.ts new file mode 100644 index 00000000000..dd9d940d9df --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-publication.test.ts @@ -0,0 +1,44 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { onNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js"; +import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; + +describe("native harness backup publication", () => { + let backupRoot = ""; + let unsubscribe: (() => void) | undefined; + afterEach(async () => { + unsubscribe?.(); unsubscribe = undefined; + if (backupRoot) await fs.rm(backupRoot, { recursive: true, force: true }); + }); + async function fixture() { + backupRoot = await fs.mkdtemp(path.join(os.tmpdir(), "native-backup-publication-")); + const current = path.join(backupRoot, "current"), pendingRoot = path.join(backupRoot, "pending"); + await fs.mkdir(current); await fs.mkdir(pendingRoot); + await fs.writeFile(path.join(current, "manifest.json"), "saved"); + await fs.writeFile(path.join(pendingRoot, "manifest.json"), "replacement"); + return { db: {}, companyId: "company", scopeDigest: "scope", backupRoot, pendingRoot, current }; + } + it("restores the existing backup when its lease stamp fails without publishing a false change", async () => { + const input = await fixture(); const changes: unknown[] = []; + unsubscribe = onNativeHarnessBackupChange(input.db, change => changes.push(change)); + await expect(publishNativeHarnessBackup({ ...input, stamp: async current => { + expect(await fs.readFile(path.join(current, "manifest.json"), "utf8")).toBe("replacement"); + throw new Error("stamp transaction rejected"); + } })).rejects.toThrow("stamp transaction rejected"); + expect(await fs.readFile(path.join(input.current, "manifest.json"), "utf8")).toBe("saved"); + expect(await fs.readdir(backupRoot)).toEqual(["current"]); expect(changes).toEqual([]); + }); + it("publishes the completed replacement once and removes its predecessor", async () => { + const input = await fixture(); const changes: unknown[] = []; + unsubscribe = onNativeHarnessBackupChange(input.db, change => changes.push(change)); + await publishNativeHarnessBackup({ ...input, stamp: async current => { + expect(changes).toEqual([]); + expect(await fs.readFile(path.join(current, "manifest.json"), "utf8")).toBe("replacement"); + } }); + expect(await fs.readFile(path.join(input.current, "manifest.json"), "utf8")).toBe("replacement"); + expect(await fs.readdir(backupRoot)).toEqual(["current"]); + expect(changes).toEqual([{ companyId: input.companyId, scopeDigest: input.scopeDigest }]); + }); +}); diff --git a/server/src/services/native-runtime/native-harness-backup-publication.ts b/server/src/services/native-runtime/native-harness-backup-publication.ts new file mode 100644 index 00000000000..820303584b0 --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-publication.ts @@ -0,0 +1,28 @@ +import { existsSync, renameSync } from "node:fs"; +import { resolve } from "node:path"; +import { noteNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js"; +import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; + +export async function publishNativeHarnessBackup(input: { + db: object; companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string; + stamp?: (currentRoot: string) => Promise; +}) { + const currentRoot = resolve(input.backupRoot, "current"); + const previousRoot = resolve(input.backupRoot, "previous"); + removeNativeHarnessBackup(previousRoot); + let movedCurrent = false; + if (existsSync(currentRoot)) { + renameSync(currentRoot, previousRoot); + movedCurrent = true; + } + try { + renameSync(input.pendingRoot, currentRoot); + await input.stamp?.(currentRoot); + } catch (error) { + if (existsSync(currentRoot)) removeNativeHarnessBackup(currentRoot); + if (movedCurrent && existsSync(previousRoot) && !existsSync(currentRoot)) renameSync(previousRoot, currentRoot); + throw error; + } + removeNativeHarnessBackup(previousRoot); + noteNativeHarnessBackupChange(input.db, { companyId: input.companyId, scopeDigest: input.scopeDigest }); +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index b47ddc40fee..894dc631cee 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,4 @@ -import { noteNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js"; +import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeTurnAccounting, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent } from "./native-usage-accounting.js"; export { nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage } from "./native-usage-accounting.js"; @@ -13277,42 +13277,15 @@ async function createRunnerdBackendWithinSessionClaim( }); renameSync(temporaryManifest, manifestPath); - const currentRoot = resolve(backupRoot, "current"); - const previousRoot = resolve(backupRoot, "previous"); - removeNativeHarnessBackup(previousRoot); - let movedCurrent = false; - if (existsSync(currentRoot)) { - renameSync(currentRoot, previousRoot); - movedCurrent = true; - } - try { - renameSync(pendingRoot, currentRoot); - if ( - remoteTarget?.transport === "sandbox" && - remoteTarget.leaseId - ) { + await publishNativeHarnessBackup({ + db: input.db, companyId: input.execution.binding.companyId, + scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex"), backupRoot, pendingRoot, + stamp: remoteTarget?.transport === "sandbox" && remoteTarget.leaseId ? async currentRoot => { await recordHarnessBackupStampForCurrentLease({ - root: currentRoot, - manifest, - bytes: backupSpanAttributes.bytesTransferred, + root: currentRoot, manifest, bytes: backupSpanAttributes.bytesTransferred, }); - } - } catch (error) { - if (existsSync(currentRoot)) { - removeNativeHarnessBackup(currentRoot); - } - if ( - movedCurrent && - existsSync(previousRoot) && - !existsSync(currentRoot) - ) { - renameSync(previousRoot, currentRoot); - } - throw error; - } - removeNativeHarnessBackup(previousRoot); - noteNativeHarnessBackupChange(input.db, { companyId: input.execution.binding.companyId, - scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex") }); + } : undefined, + }); } finally { removeNativeHarnessBackup(pendingRoot); } From 17564db697a1b6d3c1a417a0f0a719b38885b5f3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:25:54 -0500 Subject: [PATCH 3/6] fix(server): isolate backup discovery and recover orphan cleanup Co-Authored-By: Paperclip --- .../2026-10-30-backup-expiration-sleep.md | 8 ++- .../native-harness-backup-expiration.test.ts | 59 +++++++++++++++++++ .../server-startup-feedback-export.test.ts | 10 +++- .../native-harness-backup-expiration-work.ts | 37 ++++++++++-- .../native-harness-backup-expiration.ts | 19 +++--- 5 files changed, 115 insertions(+), 18 deletions(-) diff --git a/doc/plans/2026-10-30-backup-expiration-sleep.md b/doc/plans/2026-10-30-backup-expiration-sleep.md index 47d41bbf581..91cf1e9ec0a 100644 --- a/doc/plans/2026-10-30-backup-expiration-sleep.md +++ b/doc/plans/2026-10-30-backup-expiration-sleep.md @@ -14,7 +14,9 @@ must invalidate the source when it changes. A busy process or failed I/O leaves real unresolved cleanup and uses the shared retry backoff. Dry-run reports an eligible source once and waits for source invalidation or restart. -Startup pages existing disk scopes. Roots without backup or cleanup tombstones +Startup pages existing disk scopes and indexes company ownership, including +protected backups. Company notifications inspect only that index; local backup +publication adds newly created roots. Roots without backup or cleanup tombstones perform no backup SQL. After discovery and all obligations settle, there is no recurring SQL or filesystem scan. Startup reconstructs future work; inspecting a source never changes its business timestamps. Runtime task close/reopen and @@ -32,7 +34,9 @@ no permanent database listener and does not add a polling table or migration. The existing issue share locks now remain held through the retirement rename. This fences task reopen while the backup becomes unavailable. A failed database transaction restores `.retiring-*`; a process interruption is recovered from -that existing file journal at startup. After database settlement, `.expired-*` +that existing file journal at startup. Journal reconciliation runs before +identity or owner lookup, so an agent purge cannot strand expired physical +cleanup or prevent restoration of a retiring backup. After database settlement, `.expired-*` records outstanding physical removal so a partial delete retries idempotently. Ownership, session identity, live run, lease and local execution safety checks remain in the backup owner. No host-sleep blocker or expiry-on-use rule changes. diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index 1ac2539545e..afadf5d1d63 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -1,3 +1,4 @@ +import { agentService } from "../services/agents.js"; import { publishNativeHarnessBackup } from "../services/native-runtime/native-harness-backup-publication.js"; import { registerNativeHarnessBackupExpirationWork } from "../services/native-runtime/native-harness-backup-expiration-work.js"; import { noteNativeHarnessBackupChange } from "../services/native-harness-backup-expiration-notifications.js"; @@ -711,6 +712,64 @@ describeEmbeddedPostgres("native harness backup expiration", () => { await quiet(); }); + it("company-only notifications inspect only the indexed company's blocked backup sources", async () => { + const own = await seedSession({ issueStatus: "todo" }); + const other = await seedSession({ issueStatus: "todo" }); + await startBackups({ maxEntries: 1 }); await settled(); await quiet(); + const sql: Array<{ query: string; parameters: unknown[] }> = []; + const previous = db.$client.options.debug; + db.$client.options.debug = (_connection, query, parameters) => { sql.push({ query, parameters }); }; + const directories = vi.spyOn(fs, "readdir"); + try { + const closed = await issueService(db).update(own.issueId, { status: "done" }); + await settled(new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1); + const unrelated = [other.companyId, other.agentId, other.runId, other.sessionId, other.issueId]; + expect(sql.filter(query => query.parameters.some(parameter => + unrelated.some(id => String(parameter).includes(id))))).toEqual([]); + expect(directories.mock.calls.some(call => call[0] === stateBase + || String(call[0]).startsWith(other.root))).toBe(false); + expect(await exists(other.current)).toBe(true); + } finally { db.$client.options.debug = previous; directories.mockRestore(); } + await quiet(); + }); + + it.each([false, true])("finishes cleanup journals after ordinary agent deletion (restart %s)", async restart => { + const session = await seedSession(); const remove = fs.rm; + const fault = vi.spyOn(fs, "rm").mockImplementation(async (...args) => { + if (String(args[0]).includes(".expired-")) throw new Error("controlled orphan cleanup failure"); + return remove(...args); + }); + await startBackups(); + expect(backupErrors).toHaveLength(1); + const retryAt = coordinator!.nextWakeAt()!; + const backupRoot = path.dirname(session.current); + expect((await fs.readdir(backupRoot)).some(name => name.startsWith(".expired-"))).toBe(true); + if (restart) { await backupWorker!.stop(); await coordinator!.stop(); } + await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" }).where(eq(agents.id, session.agentId)); + await agentService(db).remove(session.agentId); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, session.runId))).toEqual([]); + fault.mockRestore(); + if (restart) await startBackups(); else await vi.advanceTimersByTimeAsync(retryAt - Date.now()); + await settled(); + expect(await fs.readdir(backupRoot)).toEqual([]); expect(await taskSessionCount(session.agentId)).toBe(0); + await quiet(); + }); + + it("restores an orphan retiring journal without expiring the recovered backup or unrelated names", async () => { + const session = await seedSession(); const backupRoot = path.dirname(session.current); + const retiring = path.join(backupRoot, `.retiring-${randomUUID()}`); + await fs.rename(session.current, retiring); + await fs.mkdir(path.join(backupRoot, ".expired-unrecognized")); + await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" }).where(eq(agents.id, session.agentId)); + await agentService(db).remove(session.agentId); + await fs.rm(path.join(session.root, "control-plane"), { recursive: true }); + await startBackups(); await settled(); + expect(await exists(session.current)).toBe(true); expect(await exists(retiring)).toBe(false); + expect(await exists(path.join(session.current, "manifest.json"))).toBe(true); + expect(await exists(path.join(backupRoot, ".expired-unrecognized"))).toBe(true); + await quiet(); + }); + it("retains live lease references and only removes the owning company's backup after lease release", async () => { const first = await seedSession({ leaseStatus: "retained" }); const other = await seedSession({ leaseStatus: "retained" }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index b645d7437c8..801c4c64ea6 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -566,6 +566,8 @@ vi.mock("../services/native-runtime/native-harness-backup-expiration.js", () => tick: vi.fn(async () => ({ checked: 0, candidates: 0, evidenceCandidates: 0, failed: 0, truncated: false })), noteEvidence: vi.fn(), nextEvidenceAt: vi.fn(() => null), tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn(), + backupScopes: vi.fn(async () => ({ names: [], complete: true })), + reconcileBackup: vi.fn(async () => null), expireBackup: vi.fn(), })), })); @@ -676,9 +678,12 @@ describe("startServer feedback export wiring", () => { loadConfigMock.mockReturnValue(buildTestConfig({ nativeHarnessBackupExpirationEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.nativeRunnerEvidenceRetention)).toHaveLength(enabled ? 1 : 0); - expect(vi.mocked(nativeHarnessBackupExpirationService).mock.calls.map(([, options]) => options?.duty)).toEqual(enabled ? ["backups", "evidence"] : ["backups"]); + expect(vi.mocked(nativeHarnessBackupExpirationService).mock.calls.map(([, options]) => options?.duty)).toEqual(enabled ? ["backups", "backups", "evidence"] : ["backups"]); const backups = vi.mocked(nativeHarnessBackupExpirationService).mock.results[0]!.value; expect(backups.registerCheckpointExpiration).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.nativeHarnessBackupExpiration)).toHaveLength(enabled ? 1 : 0); + expect(backups.tick).not.toHaveBeenCalled(); + expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.nativeHarnessBackupExpiration)).toEqual([]); }); it("tracks held evidence reconstruction after HTTP readiness", async () => { @@ -686,6 +691,9 @@ describe("startServer feedback export wiring", () => { const pending = Promise.withResolvers<{ truncated: boolean }>(); vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({ tick: vi.fn(), noteEvidence: vi.fn(), nextEvidenceAt: () => null, tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn() } as never); + vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({ + backupScopes: vi.fn(async () => ({ names: [], complete: true })), reconcileBackup: vi.fn(), expireBackup: vi.fn(), + } as never); vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({ tick: vi.fn(() => pending.promise), noteEvidence: vi.fn(), nextEvidenceAt: () => null, tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn() } as never); try { diff --git a/server/src/services/native-runtime/native-harness-backup-expiration-work.ts b/server/src/services/native-runtime/native-harness-backup-expiration-work.ts index 93c4f08e340..f0dafe195e0 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration-work.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration-work.ts @@ -15,6 +15,26 @@ export function registerNativeHarnessBackupExpirationWork( ) { const service = nativeHarnessBackupExpirationService(db, { ...options, duty: "backups" }); const cursors = new Map(); + const roots = new Map>(); + const owners = new Map(); + function remember(id: string, companyId?: string) { + const previous = owners.get(id); + if (previous && previous !== companyId) { + const companyRoots = roots.get(previous)!; + companyRoots.delete(id); + if (!companyRoots.size) roots.delete(previous); + owners.delete(id); + } + if (!companyId) return; + let companyRoots = roots.get(companyId); + if (!companyRoots) { companyRoots = new Set(); roots.set(companyId, companyRoots); } + companyRoots.add(id); owners.set(id, companyId); + } + async function reconcile(id: string): Promise { + const source = await service.reconcileBackup(id); + remember(id, source?.companyId); + return source && source.dueAt !== null ? { ...source, dueAt: source.dueAt } : null; + } const worker = registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.nativeHarnessBackupExpiration, batchSize: options.maxRemovals ?? 25, decodeScope: (hint: string): NativeHarnessBackupExpirationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), @@ -22,12 +42,16 @@ export function registerNativeHarnessBackupExpirationWork( && (!scope.scopeDigest || entry.id === scope.scopeDigest), load: async (scope, signal) => { const key = scope ? JSON.stringify(scope) : "*"; + const limit = options.maxEntries ?? 100; + const indexed = scope ? [...roots.get(scope.companyId) ?? []] + .filter(id => !cursors.get(key) || id > cursors.get(key)!).sort() : null; const page = scope?.scopeDigest ? { names: [scope.scopeDigest], complete: true } - : await service.backupScopes(cursors.get(key), options.maxEntries ?? 100); + : indexed ? { names: indexed.slice(0, limit), complete: indexed.length <= limit } + : await service.backupScopes(cursors.get(key), limit); const rows: Entry[] = []; for (const name of page.names) { if (signal?.aborted || !canRun()) throw new Error("Backup expiration discovery paused"); - const entry = await service.inspectBackup(name); + const entry = await reconcile(name); if (entry && (!scope || entry.companyId === scope.companyId)) rows.push(entry); } if (page.complete) cursors.delete(key); else cursors.set(key, page.names.at(-1)!); @@ -35,19 +59,22 @@ export function registerNativeHarnessBackupExpirationWork( }, check: async (entry, signal) => { if (signal?.aborted || !canRun()) throw new Error("Backup expiration paused"); - const current = await service.inspectBackup(entry.id); + const current = await reconcile(entry.id); if (!current || current.companyId !== entry.companyId) return null; if (current.dueAt > Date.now()) return current; const result = await service.expireBackup(entry.id); if (result.candidates || result.failed) logger.info({ ...result, dryRun: options.dryRun ?? false }, "native harness backup expiration"); if (options.dryRun) return null; - const remaining = await service.inspectBackup(entry.id); + const remaining = await reconcile(entry.id); if (!remaining) return null; if (result.skipped.session_active) throw new Error("Backup session is active"); return remaining; }, }, canRun); - const unsubscribe = onNativeHarnessBackupChange(db, scope => worker.invalidate(JSON.stringify(scope))); + const unsubscribe = onNativeHarnessBackupChange(db, scope => { + if (scope.scopeDigest) remember(scope.scopeDigest, scope.companyId); + worker.invalidate(JSON.stringify(scope)); + }); const release = onRetainedNativeSessionCleanupRelease(db, scope => worker.invalidate(JSON.stringify({ companyId: scope.companyId }))); return { ...worker, async stop() { unsubscribe(); release(); await worker.stop(); } }; } diff --git a/server/src/services/native-runtime/native-harness-backup-expiration.ts b/server/src/services/native-runtime/native-harness-backup-expiration.ts index 09ddea9e5d2..ad9ac18cfb2 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration.ts @@ -739,30 +739,29 @@ export function nativeHarnessBackupExpirationService( }); } - async function inspectBackup(scopeDigest: string) { + async function reconcileBackup(scopeDigest: string) { const base = path.resolve(defaults.stateBase ?? nativeHarnessStateBase()); const root = path.join(base, scopeDigest); if (!SCOPE_PATTERN.test(scopeDigest) || !(await isRealDirectory(root))) return null; const backupRoot = path.join(root, "failover-backups"); if (!(await isRealDirectory(backupRoot))) return null; - const names = await fs.readdir(backupRoot); + if (!defaults.dryRun) await settleLeftovers(backupRoot); const current = path.join(backupRoot, "current"); const stat = await fs.lstat(current).catch(missingPath); - const leftovers = names.some(name => [RETIRING_PREFIX, EXPIRED_PREFIX].some(prefix => - name.startsWith(prefix) && UUID_PATTERN.test(name.slice(prefix.length)))); - if ((!stat?.isDirectory() || stat.isSymbolicLink()) && !leftovers) return null; - if (defaults.dryRun && !stat?.isDirectory()) return null; + if (!stat?.isDirectory() || stat.isSymbolicLink()) return null; const identity = await readSessionIdentity(root); if (!identity) return null; const [owner] = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) .from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); if (!owner) return null; - if (leftovers && !defaults.dryRun) return { id: scopeDigest, ...owner, dueAt: 0 }; - if (await fs.lstat(path.join(root, "cleanup-activation.json")).then(() => true, () => false)) return null; + const source = { id: scopeDigest, ...owner }; + if (await fs.lstat(path.join(root, "cleanup-activation.json")).then(() => true, () => false)) { + return { ...source, dueAt: null }; + } const blocker = await db.transaction(tx => sessionRetentionBlocker(tx, { scopeDigest, identity, closedBefore: new Date(8_000_000_000_000_000), })); - if ("reason" in blocker) return null; + if ("reason" in blocker) return { ...source, dueAt: null }; const retentionMs = Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS; return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 }; } @@ -789,5 +788,5 @@ export function nativeHarnessBackupExpirationService( } return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration, - inspectBackup, backupScopes, expireBackup }; + reconcileBackup, backupScopes, expireBackup }; } From bdb9b66ac6d8c9e3ac81131513379e5a0d547e39 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:54:51 -0500 Subject: [PATCH 4/6] test(skills): verify Cloud CLI reads through shared object storage Use the real S3 provider with a controlled HTTP fixture and a stable test key. Preserve read-only/no-seeding behavior and verify corrupt, missing, and restored object outcomes under managed Cloud policy. Co-Authored-By: Paperclip --- .../__tests__/managed-skill-content.test.ts | 90 ++++++++++++++++--- 1 file changed, 77 insertions(+), 13 deletions(-) diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index cd4ddb8135f..59566d743e3 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; @@ -79,26 +81,88 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => { const companyId = await company(); - await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const objects = new Map(); + const server = createServer((request, response) => { + void (async () => { + const key = new URL(request.url!, "http://fixture").pathname; + if (request.method === "PUT") { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + let body = Buffer.concat(chunks); + if (request.headers["content-encoding"]?.includes("aws-chunked")) { + const decoded: Buffer[] = []; + let offset = 0; + while (offset < body.length) { + const end = body.indexOf("\r\n", offset); + const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16); + if (size === 0) break; + decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2; + } + body = Buffer.concat(decoded); + } + objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return; + } + const body = objects.get(key); + if (!body) { response.writeHead(404).end(); return; } + response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' }); + response.end(request.method === "HEAD" ? undefined : body); + })().catch(() => { response.writeHead(500).end(); }); + }); + await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address"); + const endpoint = `http://127.0.0.1:${address.port}`; + const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" }; + const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]])); + Object.assign(process.env, credentials); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({ + bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true, + }))); const key = "cloud-runtime-config:v1"; - const run = (phase: string) => promisify(execFile)(process.execPath, [ - "--import", fileURLToPath(import.meta.resolve("tsx")), - fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), - `--phase=${phase}`, `--company=${companyId}`, "--limit=50", - ], { env: { ...process.env, DATABASE_URL: database.connectionString, - PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", - PAPERCLIP_STORAGE_PROVIDER: "local_disk", PAPERCLIP_STORAGE_LOCAL_DIR: path.join(root, "objects") }, - timeout: 60_000, maxBuffer: 1024 * 1024 }); - await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); - expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); - await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); try { + await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const run = (phase: string) => promisify(execFile)(process.execPath, [ + "--import", fileURLToPath(import.meta.resolve("tsx")), + fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), + `--phase=${phase}`, `--company=${companyId}`, "--limit=50", + ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString, + PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", + PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3", + PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1", + PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" }, + timeout: 60_000, maxBuffer: 1024 * 1024 }); + await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); + await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!); expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]); + const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const); const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(verification.results).toEqual([expect.objectContaining({ status: "local_storage" })]); + expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()]).toEqual(before); + const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!; + const archive = objects.get(archiveKey)!; + const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1; + objects.set(archiveKey, corrupted); + const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]); + objects.set(archiveKey, archive); + const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!; + const manifest = objects.get(manifestKey)!; objects.delete(manifestKey); + const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]); + objects.set(manifestKey, manifest); + const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()].sort()).toEqual([...before].sort()); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))) + .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]); } finally { await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key)); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")))); + for (const [key, value] of Object.entries(previousCredentials)) { + if (value === undefined) delete process.env[key]; else process.env[key] = value; + } + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); } }, 180_000); From 6b9b421905ae3691e82626c9cf5c7f06a23ddb49 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:40:27 -0500 Subject: [PATCH 5/6] fix: accept authentication URLs drained after their result Keep validation and process retirement intact when stdout is delivered before stderr. Reproduce both valid and invalid late URLs with a real child process. Co-Authored-By: Paperclip --- .../src/drivers/acpx/antigravity-auth.test.ts | 21 ++++++++++++------- .../src/drivers/acpx/antigravity-auth.ts | 4 +++- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.test.ts b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.test.ts index 897a4d675be..af78eef13bb 100644 --- a/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.test.ts @@ -45,9 +45,9 @@ reader.on('line', line => { if(request.method==='initialize') { process.stdout.write(JSON.stringify({jsonrpc:'2.0',id:request.id,result:{protocolVersion:1,agentInfo:{name:'antigravity-acp',version:'1.4.3'},agentCapabilities:{_meta:{antigravity:{paperclipAdmission:1}}},authMethods:[{id:'oauth-personal'}]}})+'\\n'); } else if(request.method==='authenticate') { - const url = mode==='bad-url' ? 'https://example.invalid/steal' : ${JSON.stringify(authorizationUrl())}; + const url = mode.startsWith('bad-url') ? 'https://example.invalid/steal' : ${JSON.stringify(authorizationUrl())}; const message='Open the following link to authenticate the ACP server: '+url+'\\n'; - process.stderr.write(message.slice(0,42));process.stderr.write(message.slice(42)); + if (!mode.endsWith('url-after-result')) {process.stderr.write(message.slice(0,42));process.stderr.write(message.slice(42));} if(mode==='blocked' || mode==='bad-url' || mode==='stubborn' || mode==='retained') return; if(mode==='error') {process.stdout.write(JSON.stringify({jsonrpc:'2.0',id:request.id,error:{code:-32000,message:'do not expose fixture-refresh-token'}})+'\\n');return;} fs.mkdirSync(path.join(home,'antigravity-acp'),{mode:0o700}); @@ -55,7 +55,14 @@ reader.on('line', line => { process.stdout.write(JSON.stringify({jsonrpc:'2.0',id:request.id,result:{}})+'\\n'); } else {throw new Error('inference/session request forbidden');} }); -reader.on('close',()=>mode==='stubborn'||mode==='retained'?setInterval(()=>{},1000):process.exit(0)); +reader.on('close',()=> { + if (mode.endsWith('url-after-result')) { + const url = mode.startsWith('bad-url') ? 'https://example.invalid/steal' : ${JSON.stringify(authorizationUrl())}; + // EOF proves the client received the authenticate result before this stderr. + process.stderr.write('Open the following link to authenticate the ACP server: '+url+'\\n', () => process.exit(0)); + } else if(mode==='stubborn'||mode==='retained') setInterval(()=>{},1000); + else process.exit(0); +}); `); setup.verify.mockResolvedValue({ openCommand: async () => ({ spawn: setup.spawn, close: setup.close }) }); setup.spawn.mockImplementation((_args, options) => { @@ -95,8 +102,8 @@ describe("Antigravity setup-only Google authentication", () => { // The login predicate keeps its five-second bound. Allow ten more seconds // for fixture startup and the durable mocked credential checks on a busy CI worker. - it("performs native authenticate, retires it, verifies identity, and preserves refresh rotation without repeated status network calls", async () => { - const home = await fixture(), fetch = googleIdentity(); vi.stubGlobal("fetch", fetch); + it.each(["success", "url-after-result"])("performs native authenticate, retires it, verifies identity, and preserves refresh rotation without repeated status network calls (%s)", async mode => { + const home = await fixture(mode), fetch = googleIdentity(); vi.stubGlobal("fetch", fetch); const login = startAntigravityBrowserLogin(home); await vi.waitFor(() => expect(login.outcome).toBe("success"), { timeout: 5_000 }); expect(login.authorizationUrl).toBe(authorizationUrl()); @@ -118,13 +125,13 @@ describe("Antigravity setup-only Google authentication", () => { expect(fetch).toHaveBeenCalledTimes(6); }, 15_000); - it.each(["error", "bad-url"])("rejects %s without leaking provider text or calling Google", async mode => { + it.each(["error", "bad-url", "bad-url-after-result"])("rejects %s without leaking provider text or calling Google", async mode => { const home = await fixture(mode), fetch = vi.fn(); vi.stubGlobal("fetch", fetch); const login = startAntigravityBrowserLogin(home); await vi.waitFor(() => expect(login.outcome).toBe("failure"), { timeout: 5_000 }); await vi.waitFor(() => expect(setup.close).toHaveBeenCalledOnce(), { timeout: 5_000 }); expect(login.error).not.toContain("fixture"); - if (mode === "bad-url") expect(login.authorizationUrl).toBeUndefined(); + if (mode.startsWith("bad-url")) expect(login.authorizationUrl).toBeUndefined(); expect(fetch).not.toHaveBeenCalled(); }); diff --git a/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts index 2921fd141db..2561be31d56 100644 --- a/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts +++ b/packages/paperclip-runner/src/drivers/acpx/antigravity-auth.ts @@ -203,7 +203,9 @@ export function startAntigravityBrowserLogin(home: string): AntigravityBrowserLo if (!line.startsWith(AUTH_PREFIX)) continue; try { const url = validateAntigravityAuthorizationUrl(line.slice(AUTH_PREFIX.length)); - if (phase !== 2 || state.authorizationUrl !== undefined) throw new Error(LOGIN_ERROR); + // stdout and stderr are separate pipes: the authenticate result + // can arrive before its authorization URL is drained from stderr. + if ((phase !== 2 && phase !== 3) || state.authorizationUrl !== undefined) throw new Error(LOGIN_ERROR); state.authorizationUrl = url; } catch { fail(); } } From 259a3ae4dfd5d837c0307615e97a51272eecd5a1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:01:35 -0500 Subject: [PATCH 6/6] fix(server): scope backup invalidation to retained session references Co-Authored-By: Paperclip --- .../2026-10-30-backup-expiration-sleep.md | 15 +- .../native-harness-backup-expiration.test.ts | 160 +++++++++++++++++- server/src/services/environments.ts | 4 +- server/src/services/issues.ts | 2 +- ...ve-environment-settlement-notifications.ts | 4 +- ...harness-backup-expiration-notifications.ts | 11 +- .../native-harness-backup-expiration-work.ts | 111 ++++++++---- .../native-harness-backup-expiration.ts | 76 +++++++-- .../native-runtime/native-session-executor.ts | 11 +- ...ed-native-session-cleanup-notifications.ts | 11 +- 10 files changed, 335 insertions(+), 70 deletions(-) diff --git a/doc/plans/2026-10-30-backup-expiration-sleep.md b/doc/plans/2026-10-30-backup-expiration-sleep.md index 91cf1e9ec0a..e5b3132fbc8 100644 --- a/doc/plans/2026-10-30-backup-expiration-sleep.md +++ b/doc/plans/2026-10-30-backup-expiration-sleep.md @@ -14,9 +14,13 @@ must invalidate the source when it changes. A busy process or failed I/O leaves real unresolved cleanup and uses the shared retry backoff. Dry-run reports an eligible source once and waits for source invalidation or restart. -Startup pages existing disk scopes and indexes company ownership, including -protected backups. Company notifications inspect only that index; local backup -publication adds newly created roots. Roots without backup or cleanup tombstones +Startup pages existing disk scopes and indexes each backup's actual session, +run, coalesced native-task/task, agent and lease references, including protected +backups. Ordinary task, run, lease and process-release notifications retain +these identities and inspect only affected roots. New run or lease references +resolve through the owner's database relationships. Local publication retains +its exact digest. Explicit company or instance configuration changes can still +reconstruct their whole affected scope. Roots without backup or cleanup tombstones perform no backup SQL. After discovery and all obligations settle, there is no recurring SQL or filesystem scan. Startup reconstructs future work; inspecting a source never changes its business timestamps. Runtime task close/reopen and @@ -31,6 +35,11 @@ manual disk edits require reconstruction on restart. Database writer ingress continues to use the existing transaction-aware work signals. This change adds no permanent database listener and does not add a polling table or migration. +Read-only reconstruction does not lock issue rows. Transient identity I/O errors +retain a retry obligation; missing or malformed state remains unrecognized. +Byte accounting occurs before the retirement transaction. The current backup +age is rechecked inside the transaction before the rename. + The existing issue share locks now remain held through the retirement rename. This fences task reopen while the backup becomes unavailable. A failed database transaction restores `.retiring-*`; a process interruption is recovered from diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index afadf5d1d63..9dddca536f3 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -1,3 +1,6 @@ +import { notifyNativeEnvironmentSettlement } from "../services/native-environment-settlement-notifications.js"; +import { notifyRetainedNativeSessionCleanup, onRetainedNativeSessionCleanupRelease, releaseRetainedNativeSessionCleanup } from "../services/retained-native-session-cleanup-notifications.js"; +import { environmentService } from "../services/environments.js"; import { agentService } from "../services/agents.js"; import { publishNativeHarnessBackup } from "../services/native-runtime/native-harness-backup-publication.js"; import { registerNativeHarnessBackupExpirationWork } from "../services/native-runtime/native-harness-backup-expiration-work.js"; @@ -14,7 +17,7 @@ import os from "node:os"; import path from "node:path"; import { promises as fs } from "node:fs"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; import { agents, agentTaskSessions, @@ -84,6 +87,7 @@ describeEmbeddedPostgres("native harness backup expiration", () => { }); async function seedSession(input: { + companyId?: string; agentId?: string; issueStatus?: string; closedDaysAgo?: number; runStatus?: string; @@ -93,9 +97,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => { cleanupActive?: boolean; } = {}) { if (!stateBase) stateBase = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-durable-sessions-")); - const companyId = randomUUID(); - await db.insert(companies).values({ id: companyId, name: "Backups", issuePrefix: `B${companyId.slice(0, 6)}` }); - const [agent] = await db.insert(agents).values({ companyId, name: "Coder" }).returning(); + const companyId = input.companyId ?? randomUUID(); + if (!input.companyId) await db.insert(companies).values({ id: companyId, name: "Backups", issuePrefix: `B${companyId.slice(0, 6)}` }); + const [agent] = input.agentId ? [{ id: input.agentId }] + : await db.insert(agents).values({ companyId, name: "Coder" }).returning(); const issueStatus = input.issueStatus ?? "done"; const closedAt = new Date(now.getTime() - (input.closedDaysAgo ?? 10) * DAY_MS); const [issue] = await db.insert(issues).values({ @@ -733,6 +738,153 @@ describeEmbeddedPostgres("native harness backup expiration", () => { await quiet(); }); + it("coalesces repeated task writes to one retained root in a large company with the same historical agent", async () => { + const own = await seedSession({ issueStatus: "todo" }); + const unrelated = []; + for (let index = 0; index < 80; index++) unrelated.push(await seedSession({ companyId: own.companyId, + agentId: own.agentId, issueStatus: index % 2 ? "todo" : "done", closedDaysAgo: 2, + runStatus: index % 3 ? "succeeded" : "running" })); + const due = now.getTime() + 5 * DAY_MS + 1; + await startBackups({ isSessionBusy: digest => unrelated.some((entry, index) => index % 4 === 0 && entry.scope === digest) }); + await settled(due); await quiet(60_000); + const reads = vi.spyOn(fs, "readFile"); + const queries: Array<{ query: string; parameters: unknown[] }> = []; + const previous = db.$client.options.debug; + db.$client.options.debug = (_connection, query, parameters) => { queries.push({ query, parameters }); }; + try { + await db.transaction(async tx => { + await issueService(db).update(own.issueId, { status: "done" }, tx); + for (let index = 0; index < 12; index++) await issueService(db).update(own.issueId, + { completedAt: new Date(now.getTime() - DAY_MS) }, tx); + }); + const writes = queries.length; + await settled(due); + expect(reads.mock.calls.map(call => String(call[0]))).toEqual([path.join(own.root, "control-plane", "control-plane-state.json")]); + expect(queries.length - writes).toBeLessThanOrEqual(20); + const otherIds = unrelated.flatMap(entry => [entry.issueId, entry.runId, entry.sessionId]); + expect(queries.filter(query => query.parameters.some(parameter => otherIds.some(id => String(parameter).includes(id))))).toEqual([]); + } finally { reads.mockRestore(); db.$client.options.debug = previous; } + await issueService(db).update(own.issueId, { status: "todo" }); await settled(due); + await quiet(); expect(await exists(own.current)).toBe(true); + }); + + it("indexes protected session relations so a new run, historical task close and exact process release reactivate only its backup", async () => { + const own = await seedSession({ closedDaysAgo: 2 }); + const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" }); + const [linked] = await db.insert(issues).values({ companyId: own.companyId, title: "New linked task", status: "todo" }).returning(); + const due = now.getTime() + 5 * DAY_MS + 1; + await startBackups(); await settled(due); await quiet(60_000); + const reads = vi.spyOn(fs, "readFile"); + const runId = randomUUID(); + try { + await db.transaction(async tx => { + await notifyNativeEnvironmentSettlement(tx, { companyId: own.companyId, runId, issueId: linked!.id }); + await tx.insert(heartbeatRuns).values({ id: runId, companyId: own.companyId, agentId: own.agentId, + nativeSessionId: own.sessionId, nativeIssueId: linked!.id, issueId: other.issueId, status: "running" }); + }); + await settled(); + expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true); + expect(reads).toHaveBeenCalled(); reads.mockClear(); + await issueService(db).update(linked!.id, { status: "done" }); await settled(); + expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true); reads.mockClear(); + await db.transaction(async tx => { + await notifyRetainedNativeSessionCleanup(tx, { companyId: own.companyId, runId, agentId: own.agentId }); + await tx.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + }); + const [closed] = await db.select().from(issues).where(eq(issues.id, linked!.id)); + const later = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1; + await settled(later); reads.mockClear(); + const cleanupScope = { companyId: own.companyId, agentId: own.agentId }; + const observed = vi.fn(); const stop = onRetainedNativeSessionCleanupRelease(db, observed); + releaseRetainedNativeSessionCleanup(db, cleanupScope, { companyId: own.companyId, runId }); + expect(observed).toHaveBeenCalledWith(cleanupScope, { companyId: own.companyId, runId }); stop(); + await settled(later); + expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true); + expect(reads).toHaveBeenCalled(); + await issueService(db).update(linked!.id, { status: "todo" }); await settled(); + } finally { reads.mockRestore(); } + expect(await exists(other.current)).toBe(true); await quiet(); + }); + + it("ordinary lease release without a run uses its source identity instead of other retained roots", async () => { + const own = await seedSession({ leaseStatus: "retained" }); + const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" }); + await startBackups(); await settled(); await quiet(60_000); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.companyId, own.companyId)); + const reads = vi.spyOn(fs, "readFile"); + try { + await environmentService(db).releaseLease(lease!.id, "expired"); await settled(); + expect(reads).toHaveBeenCalled(); + expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true); + } finally { reads.mockRestore(); } + expect(await exists(own.current)).toBe(false); expect(await exists(other.current)).toBe(true); await quiet(); + }); + + it("retains the future deadline when a lease metadata edit removes its last backup reference", async () => { + const own = await seedSession({ leaseStatus: "retained", closedDaysAgo: 2 }); + const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" }); + await startBackups(); await settled(); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.companyId, own.companyId)); + const reads = vi.spyOn(fs, "readFile"); const due = now.getTime() + 5 * DAY_MS + 1; + try { + await environmentService(db).updateLeaseMetadata(lease!.id, {}); await settled(due); + expect(reads).toHaveBeenCalled(); expect(reads.mock.calls.every(call => String(call[0]).startsWith(own.root))).toBe(true); + } finally { reads.mockRestore(); } + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(own.current)).toBe(false); expect(await exists(other.current)).toBe(true); await quiet(); + }); + + it("allows an ordinary task reopen while backup byte counting is held outside the retirement lock", async () => { + const own = await seedSession(); const lstat = fs.lstat; + const entered = Promise.withResolvers(), release = Promise.withResolvers(); + const held = vi.spyOn(fs, "lstat").mockImplementation(async (...args) => { + if (String(args[0]) === path.join(own.current, "codex-home", "history.jsonl")) { + entered.resolve(); await release.promise; + } + return lstat(...args); + }); + const startup = startBackups(); await entered.promise; + try { + await db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout = '500ms'`); + await issueService(db).update(own.issueId, { status: "todo" }, tx); + }); + } finally { release.resolve(); await startup; held.mockRestore(); } + await settled(); expect(await exists(own.current)).toBe(true); expect(backupErrors).toEqual([]); await quiet(); + }); + + it.each(["EIO", "EACCES"])("retains a failed identity read as real retry debt (%s)", async code => { + const own = await seedSession(); const read = fs.readFile; + const fault = vi.spyOn(fs, "readFile").mockImplementation(async (...args) => { + if (String(args[0]) === path.join(own.root, "control-plane", "control-plane-state.json")) { + throw Object.assign(new Error("controlled identity I/O failure"), { code }); + } + return read(...args); + }); + await startBackups(); expect(backupErrors).toHaveLength(1); + const retry = coordinator!.nextWakeAt()!; expect(retry - Date.now()).toBeGreaterThanOrEqual(29_000); + expect(await exists(own.current)).toBe(true); fault.mockRestore(); await quiet(5000); + await vi.advanceTimersByTimeAsync(retry - Date.now()); await settled(); + expect(await exists(own.current)).toBe(false); expect(await taskSessionCount(own.agentId)).toBe(0); await quiet(); + }); + + it("reconstructs future retention without issue row locks", async () => { + await seedSession({ closedDaysAgo: 2 }); const queries: string[] = []; + const previous = db.$client.options.debug; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { + await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1); + expect(queries.some(query => /for\s+share/i.test(query))).toBe(false); + } finally { db.$client.options.debug = previous; } + await quiet(); + }); + + it("rejects destructive backup scopes outside the controller digest namespace", async () => { + await emptyStateBase(); const remove = vi.spyOn(fs, "rm"); + await expect(service().expireBackup("../outside")).rejects.toThrow("Invalid native harness backup scope"); + expect(remove).not.toHaveBeenCalled(); remove.mockRestore(); + }); + it.each([false, true])("finishes cleanup journals after ordinary agent deletion (restart %s)", async restart => { const session = await seedSession(); const remove = fs.rm; const fault = vi.spyOn(fs, "rm").mockImplementation(async (...args) => { diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index ea72243e464..831ab854f04 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1701,7 +1701,7 @@ export function environmentService(db: Db) { cleanupStatus: environmentLeases.cleanupStatus, releasedAt: environmentLeases.releasedAt, }).from(environmentLeases).where(condition).for("update"); if (!previous) return null; - await notifyEnvironmentCleanup(tx, { companyId: previous.companyId, runId: previous.heartbeatRunId ?? undefined }); + await notifyEnvironmentCleanup(tx, { companyId: previous.companyId, runId: previous.heartbeatRunId ?? undefined, leaseId: id }); if (previous.heartbeatRunId && (previous.status !== status || (options?.cleanupStatus !== undefined && previous.cleanupStatus !== options.cleanupStatus) || (status !== "retained" && !previous.releasedAt))) { @@ -1821,7 +1821,7 @@ export function environmentService(db: Db) { const [lease] = await tx.select({ companyId: environmentLeases.companyId, heartbeatRunId: environmentLeases.heartbeatRunId, metadata: environmentLeases.metadata }) .from(environmentLeases).where(eq(environmentLeases.id, id)).for("update"); if (!lease) return null; - await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); + await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined, leaseId: id }); return tx.update(environmentLeases) .set({ metadata: preserveReusableLeaseHandoff(lease.metadata, withoutCleanupVerification(metadata)), diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index fdb02c5ea9e..6b654a701f8 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -11488,7 +11488,7 @@ export function issueService(db: Db) { patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number; } if (patch.status !== undefined || patch.completedAt !== undefined || patch.cancelledAt !== undefined) { - await notifyNativeHarnessBackupExpiration(tx, { companyId: receiptExisting.companyId }); + await notifyNativeHarnessBackupExpiration(tx, { companyId: receiptExisting.companyId, issueId: id }); } if ((patch.status !== undefined && patch.status !== receiptExisting.status) || patch.completedAt !== undefined || patch.cancelledAt !== undefined diff --git a/server/src/services/native-environment-settlement-notifications.ts b/server/src/services/native-environment-settlement-notifications.ts index 427c43522b7..0c55810b6a8 100644 --- a/server/src/services/native-environment-settlement-notifications.ts +++ b/server/src/services/native-environment-settlement-notifications.ts @@ -2,9 +2,9 @@ import { notifyNativeHarnessBackupExpiration } from "./native-harness-backup-exp import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string }; +export type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string; leaseId?: string }; export async function notifyNativeEnvironmentSettlement(tx: object, scope?: NativeEnvironmentSettlementScope) { - await notifyNativeHarnessBackupExpiration(tx, scope && { companyId: scope.companyId }); + await notifyNativeHarnessBackupExpiration(tx, scope); if (scope) await notifySkillVersionRetention(tx, scope.companyId); await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeEnvironmentSettlement, scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"); diff --git a/server/src/services/native-harness-backup-expiration-notifications.ts b/server/src/services/native-harness-backup-expiration-notifications.ts index 602a7063acd..e334487a82c 100644 --- a/server/src/services/native-harness-backup-expiration-notifications.ts +++ b/server/src/services/native-harness-backup-expiration-notifications.ts @@ -1,9 +1,16 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export type NativeHarnessBackupExpirationScope = { companyId: string; scopeDigest?: string }; +export type NativeHarnessBackupExpirationScope = { + companyId: string; scopeDigest?: string; leaseId?: string; runId?: string; issueId?: string; agentId?: string; sessionId?: string; +}; + +export function nativeHarnessBackupScopeHint(scope: NativeHarnessBackupExpirationScope) { + return JSON.stringify({ companyId: scope.companyId, scopeDigest: scope.scopeDigest, leaseId: scope.leaseId, + runId: scope.runId, issueId: scope.issueId, agentId: scope.agentId, sessionId: scope.sessionId }); +} export function notifyNativeHarnessBackupExpiration(tx: object, scope?: NativeHarnessBackupExpirationScope) { - return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeHarnessBackupExpiration, scope ? JSON.stringify(scope) : "*"); + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeHarnessBackupExpiration, scope ? nativeHarnessBackupScopeHint(scope) : "*"); } const listeners = new WeakMap void>>(); diff --git a/server/src/services/native-runtime/native-harness-backup-expiration-work.ts b/server/src/services/native-runtime/native-harness-backup-expiration-work.ts index f0dafe195e0..174f804b3a5 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration-work.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration-work.ts @@ -2,12 +2,14 @@ import type { Db } from "@paperclipai/db"; import { logger } from "../../middleware/logger.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; -import { onNativeHarnessBackupChange, type NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js"; +import { nativeHarnessBackupScopeHint, onNativeHarnessBackupChange, type NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js"; import { onRetainedNativeSessionCleanupRelease } from "../retained-native-session-cleanup-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; -import { nativeHarnessBackupExpirationService, type NativeHarnessBackupExpirationOptions } from "./native-harness-backup-expiration.js"; +import { nativeHarnessBackupExpirationService, type NativeHarnessBackupDependencies, type NativeHarnessBackupExpirationOptions } from "./native-harness-backup-expiration.js"; -type Entry = { id: string; companyId: string; agentId: string; dueAt: number }; +type Dependencies = { companyId: string; agentId: string; sessionId: string; runIds: Set; issueIds: Set; leaseIds: Set }; +type Entry = { id: string; companyId: string; agentId: string; dueAt: number; dependencies: Dependencies }; +type Reference = "company" | "agent" | "session" | "run" | "issue" | "lease"; export function registerNativeHarnessBackupExpirationWork( coordinator: ReturnType, db: Db, @@ -15,46 +17,94 @@ export function registerNativeHarnessBackupExpirationWork( ) { const service = nativeHarnessBackupExpirationService(db, { ...options, duty: "backups" }); const cursors = new Map(); - const roots = new Map>(); - const owners = new Map(); - function remember(id: string, companyId?: string) { - const previous = owners.get(id); - if (previous && previous !== companyId) { - const companyRoots = roots.get(previous)!; - companyRoots.delete(id); - if (!companyRoots.size) roots.delete(previous); - owners.delete(id); + const sources = new Map(); + const index = new Map>(); + const key = (companyId: string, kind: Reference, id = "") => JSON.stringify([companyId, kind, id]); + const sessionKey = (agentId: string, sessionId: string) => JSON.stringify([agentId, sessionId]); + function references(source: Dependencies) { + return [key(source.companyId, "company"), key(source.companyId, "agent", source.agentId), + key(source.companyId, "session", sessionKey(source.agentId, source.sessionId)), + ...[...source.runIds].map(id => key(source.companyId, "run", id)), + ...[...source.issueIds].map(id => key(source.companyId, "issue", id)), + ...[...source.leaseIds].map(id => key(source.companyId, "lease", id))]; + } + function add(reference: string, id: string) { + let roots = index.get(reference); + if (!roots) { roots = new Set(); index.set(reference, roots); } + roots.add(id); + } + function remember(id: string, current?: NativeHarnessBackupDependencies) { + const previous = sources.get(id); + if (previous) for (const reference of references(previous)) { + const roots = index.get(reference)!; + roots.delete(id); if (!roots.size) index.delete(reference); + } + if (!current) { sources.delete(id); return undefined; } + const source = { companyId: current.companyId, agentId: current.agentId, + sessionId: current.sessionId, runIds: new Set(current.runIds), issueIds: new Set(current.issueIds), leaseIds: new Set(current.leaseIds) }; + sources.set(id, source); + for (const reference of references(source)) add(reference, id); + return source; + } + function selected(scope: NativeHarnessBackupExpirationScope): Set { + if (scope.scopeDigest) return new Set([scope.scopeDigest]); + const reference = scope.leaseId ? key(scope.companyId, "lease", scope.leaseId) + : scope.runId ? key(scope.companyId, "run", scope.runId) + : scope.sessionId && scope.agentId ? key(scope.companyId, "session", sessionKey(scope.agentId, scope.sessionId)) + : scope.issueId ? key(scope.companyId, "issue", scope.issueId) + : scope.agentId ? key(scope.companyId, "agent", scope.agentId) : key(scope.companyId, "company"); + return new Set(index.get(reference) ?? []); + } + async function discoverReferences(scope: NativeHarnessBackupExpirationScope) { + if (scope.scopeDigest || (!scope.runId && !scope.leaseId)) return; + const resolved = await service.resolveBackupScope(scope); + const roots = selected(scope); + if (resolved.scopeDigest) roots.add(resolved.scopeDigest); + const related = resolved.sessionId && resolved.agentId + ? selected({ companyId: scope.companyId, sessionId: resolved.sessionId, agentId: resolved.agentId }) + : resolved.issueId ? selected({ companyId: scope.companyId, issueId: resolved.issueId }) : []; + for (const id of related) roots.add(id); + for (const id of roots) { + const source = sources.get(id); + if (!source || source.companyId !== scope.companyId) continue; + if (scope.runId) { source.runIds.add(scope.runId); add(key(scope.companyId, "run", scope.runId), id); } + if (scope.leaseId) { source.leaseIds.add(scope.leaseId); add(key(scope.companyId, "lease", scope.leaseId), id); } } - if (!companyId) return; - let companyRoots = roots.get(companyId); - if (!companyRoots) { companyRoots = new Set(); roots.set(companyId, companyRoots); } - companyRoots.add(id); owners.set(id, companyId); } async function reconcile(id: string): Promise { const source = await service.reconcileBackup(id); - remember(id, source?.companyId); - return source && source.dueAt !== null ? { ...source, dueAt: source.dueAt } : null; + const dependencies = remember(id, source?.dependencies); + return source && dependencies && source.dueAt !== null ? { ...source, dependencies, dueAt: source.dueAt } : null; + } + function matches(entry: Entry, scope?: NativeHarnessBackupExpirationScope) { + if (!scope) return true; + if (entry.companyId !== scope.companyId) return false; + if (scope.scopeDigest) return entry.id === scope.scopeDigest; + if (scope.leaseId) return entry.dependencies.leaseIds.has(scope.leaseId); + if (scope.runId) return entry.dependencies.runIds.has(scope.runId); + if (scope.sessionId) return entry.dependencies.sessionId === scope.sessionId + && (!scope.agentId || entry.agentId === scope.agentId); + if (scope.issueId) return entry.dependencies.issueIds.has(scope.issueId); + return !scope.agentId || entry.agentId === scope.agentId; } const worker = registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.nativeHarnessBackupExpiration, batchSize: options.maxRemovals ?? 25, decodeScope: (hint: string): NativeHarnessBackupExpirationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), - matches: (entry: Entry, scope?: NativeHarnessBackupExpirationScope) => !scope || entry.companyId === scope.companyId - && (!scope.scopeDigest || entry.id === scope.scopeDigest), + matches, load: async (scope, signal) => { - const key = scope ? JSON.stringify(scope) : "*"; + const hint = scope ? nativeHarnessBackupScopeHint(scope) : "*"; + if (scope && !cursors.has(hint)) await discoverReferences(scope); const limit = options.maxEntries ?? 100; - const indexed = scope ? [...roots.get(scope.companyId) ?? []] - .filter(id => !cursors.get(key) || id > cursors.get(key)!).sort() : null; - const page = scope?.scopeDigest ? { names: [scope.scopeDigest], complete: true } - : indexed ? { names: indexed.slice(0, limit), complete: indexed.length <= limit } - : await service.backupScopes(cursors.get(key), limit); + const indexed = scope ? [...selected(scope)].filter(id => !cursors.get(hint) || id > cursors.get(hint)!).sort() : null; + const page = indexed ? { names: indexed.slice(0, limit), complete: indexed.length <= limit } + : await service.backupScopes(cursors.get(hint), limit); const rows: Entry[] = []; for (const name of page.names) { if (signal?.aborted || !canRun()) throw new Error("Backup expiration discovery paused"); const entry = await reconcile(name); if (entry && (!scope || entry.companyId === scope.companyId)) rows.push(entry); } - if (page.complete) cursors.delete(key); else cursors.set(key, page.names.at(-1)!); + if (page.complete) cursors.delete(hint); else cursors.set(hint, page.names.at(-1)!); return { rows, complete: page.complete }; }, check: async (entry, signal) => { @@ -71,10 +121,7 @@ export function registerNativeHarnessBackupExpirationWork( return remaining; }, }, canRun); - const unsubscribe = onNativeHarnessBackupChange(db, scope => { - if (scope.scopeDigest) remember(scope.scopeDigest, scope.companyId); - worker.invalidate(JSON.stringify(scope)); - }); - const release = onRetainedNativeSessionCleanupRelease(db, scope => worker.invalidate(JSON.stringify({ companyId: scope.companyId }))); + const unsubscribe = onNativeHarnessBackupChange(db, scope => worker.invalidate(nativeHarnessBackupScopeHint(scope))); + const release = onRetainedNativeSessionCleanupRelease(db, (_scope, source) => worker.invalidate(nativeHarnessBackupScopeHint(source))); return { ...worker, async stop() { unsubscribe(); release(); await worker.stop(); } }; } diff --git a/server/src/services/native-runtime/native-harness-backup-expiration.ts b/server/src/services/native-runtime/native-harness-backup-expiration.ts index ad9ac18cfb2..b1d8c7c6f1d 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration.ts @@ -17,6 +17,7 @@ import { nativeHarnessStateBase } from "./native-harness-backup-stamp.js"; import { nativeSessionScopeDigestIsBusy } from "./native-session-executor.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js"; export const NATIVE_HARNESS_BACKUP_EXPIRATION_INTERVAL_MS = 60 * 60 * 1000; export const NATIVE_HARNESS_BACKUP_EXPIRATION_STARTUP_DELAY_MS = 2 * 60 * 1000; @@ -146,7 +147,9 @@ async function readSessionIdentity(root: string): Promise MAX_CONTROL_PLANE_STATE_BYTES) return null; const state = JSON.parse(await fs.readFile(statePath, "utf8")) as { identity?: Record }; + if (!state || typeof state !== "object" || Array.isArray(state)) return null; const identity = state.identity ?? {}; + if (!identity || typeof identity !== "object" || Array.isArray(identity)) return null; const runId = identity.runId; const normalizedSessionId = identity.normalizedSessionId; const environmentLeaseId = identity.environmentLeaseId; @@ -160,16 +163,21 @@ async function readSessionIdentity(root: string): Promise[0]>[0]; async function sessionRetentionBlocker( tx: Tx, - input: { scopeDigest: string | null; identity: SessionIdentity; closedBefore: Date; checkpoint?: boolean }, + input: { scopeDigest: string | null; identity: SessionIdentity; closedBefore: Date; checkpoint?: boolean; lock?: boolean }, ): Promise<{ reason: string } | { companyId: string; agentId: string; runIds: string[]; issueIds: string[]; closedAt: number }> { const [owner] = await tx .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) @@ -197,7 +205,7 @@ async function sessionRetentionBlocker( if (input.checkpoint && runs.some(run => run.finishedAt! >= input.closedBefore)) return { reason: "recent_run" }; const issueIds = [...new Set(runs.map((run) => run.issueId).filter((id): id is string => Boolean(id)))]; if (issueIds.length === 0) return { reason: "issue_unknown" }; - const issueRows = await tx + const issueQuery = tx .select({ id: issues.id, status: issues.status, @@ -205,8 +213,8 @@ async function sessionRetentionBlocker( closedAt: sql`coalesce(${issues.completedAt}, ${issues.cancelledAt})`, }) .from(issues) - .where(and(eq(issues.companyId, owner.companyId), inArray(issues.id, issueIds))) - .for("share"); + .where(and(eq(issues.companyId, owner.companyId), inArray(issues.id, issueIds))); + const issueRows = await (input.lock === false ? issueQuery : issueQuery.for("share")); if (issueRows.length !== issueIds.length) return { reason: "issue_unknown" }; for (const issue of issueRows) { if (!CLOSED_ISSUE_STATUSES.includes(issue.status)) return { reason: "issue_open" }; @@ -246,7 +254,7 @@ async function sessionRetentionBlocker( .where(and(inArray(environmentLeases.status, LIVE_LEASE_STATUSES), or(...leaseReferences))) .limit(1); if (lease) return { reason: "live_lease" }; - return { ...owner, runIds, issueIds, closedAt: Math.max(...issueRows.map(issue => new Date(issue.closedAt!).getTime())) }; + return { ...owner, runIds, issueIds, closedAt: issueRows.reduce((latest, issue) => Math.max(latest, new Date(issue.closedAt!).getTime()), Number.NEGATIVE_INFINITY) }; } async function hasLiveRun(db: Db, owner: { companyId: string; agentId: string }, sessionId: string) { @@ -379,6 +387,7 @@ export function nativeHarnessBackupExpirationService( const identity = await readSessionIdentity(root); if (!identity) return skip("identity_unknown"); + const bytes = await allocatedBytes(current); const id = randomUUID(); const retiring = path.join(backupRoot, `${RETIRING_PREFIX}${id}`); const trash = path.join(backupRoot, `${EXPIRED_PREFIX}${id}`); @@ -390,7 +399,7 @@ export function nativeHarnessBackupExpirationService( scopeDigest, identity, closedBefore: options.closedBefore, }); if ("reason" in blocker) return blocker; - const bytes = await allocatedBytes(current); + if ((await fs.stat(current)).mtime >= options.closedBefore) return { reason: "recent_backup" }; if (options.dryRun) return { cleared: 0, bytes }; if (options.isSessionBusy(scopeDigest)) return { reason: "session_active" }; await notifyExternalConversationSettlement(tx, { companyId: blocker.companyId }); @@ -739,6 +748,43 @@ export function nativeHarnessBackupExpirationService( }); } + async function backupDependencies(scopeDigest: string, identity: SessionIdentity): Promise { + const [owner] = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + issueId: sql`coalesce(${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId})`, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); + if (!owner) return null; + const runs = await db.select({ id: heartbeatRuns.id, + issueId: sql`coalesce(${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId})`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId), + eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.nativeSessionId, identity.normalizedSessionId))); + const runIds = [...new Set([identity.runId, ...runs.map(run => run.id)])]; + const issueIds = [...new Set([owner.issueId, ...runs.map(run => run.issueId)].filter((id): id is string => Boolean(id)))]; + const leases = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, owner.companyId), or(inArray(environmentLeases.heartbeatRunId, runIds), + sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'sessionScopeSha256' = ${scopeDigest}`, + identity.environmentLeaseId ? eq(environmentLeases.id, identity.environmentLeaseId) : undefined))); + return { companyId: owner.companyId, agentId: owner.agentId, sessionId: identity.normalizedSessionId, + runIds, issueIds, leaseIds: leases.map(lease => lease.id) }; + } + + async function resolveBackupScope(scope: NativeHarnessBackupExpirationScope) { + let runId = scope.runId; + let issueId = scope.issueId; + let scopeDigest = scope.scopeDigest; + if (scope.leaseId) { + const [lease] = await db.select({ runId: environmentLeases.heartbeatRunId, issueId: environmentLeases.issueId, + scopeDigest: sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'sessionScopeSha256'`, + }).from(environmentLeases).where(and(eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId))); + runId = lease?.runId ?? runId; issueId = lease?.issueId ?? issueId; + scopeDigest = lease?.scopeDigest && SCOPE_PATTERN.test(lease.scopeDigest) ? lease.scopeDigest : scopeDigest; + } + const [run] = runId ? await db.select({ agentId: heartbeatRuns.agentId, sessionId: heartbeatRuns.nativeSessionId, + issueId: sql`coalesce(${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId})`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.id, runId))) : []; + return { runId, issueId: run?.issueId ?? issueId, agentId: run?.agentId ?? scope.agentId, + sessionId: run?.sessionId ?? scope.sessionId, scopeDigest }; + } + async function reconcileBackup(scopeDigest: string) { const base = path.resolve(defaults.stateBase ?? nativeHarnessStateBase()); const root = path.join(base, scopeDigest); @@ -751,19 +797,18 @@ export function nativeHarnessBackupExpirationService( if (!stat?.isDirectory() || stat.isSymbolicLink()) return null; const identity = await readSessionIdentity(root); if (!identity) return null; - const [owner] = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) - .from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); - if (!owner) return null; - const source = { id: scopeDigest, ...owner }; + const dependencies = await backupDependencies(scopeDigest, identity); + if (!dependencies) return null; + const source = { id: scopeDigest, companyId: dependencies.companyId, agentId: dependencies.agentId, dependencies }; if (await fs.lstat(path.join(root, "cleanup-activation.json")).then(() => true, () => false)) { return { ...source, dueAt: null }; } const blocker = await db.transaction(tx => sessionRetentionBlocker(tx, { - scopeDigest, identity, closedBefore: new Date(8_000_000_000_000_000), + scopeDigest, identity, closedBefore: new Date(8_000_000_000_000_000), lock: false, })); if ("reason" in blocker) return { ...source, dueAt: null }; const retentionMs = Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS; - return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 }; + return { ...source, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 }; } async function backupScopes(after?: string, limit = 100) { @@ -776,6 +821,7 @@ export function nativeHarnessBackupExpirationService( } async function expireBackup(scopeDigest: string) { + if (!SCOPE_PATTERN.test(scopeDigest)) throw new Error("Invalid native harness backup scope"); const result = emptyResult(); const root = path.join(path.resolve(defaults.stateBase ?? nativeHarnessStateBase()), scopeDigest); const dryRun = defaults.dryRun ?? false; @@ -788,5 +834,5 @@ export function nativeHarnessBackupExpirationService( } return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration, - reconcileBackup, backupScopes, expireBackup }; + reconcileBackup, backupScopes, expireBackup, resolveBackupScope }; } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 99365030f3d..2424f60d26c 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -8092,7 +8092,8 @@ export async function executePaperclipNativeSession(input: { } releaseRetainedNativeSessionCleanup(input.db, { companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, - }); + }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId, + issueId: input.execution.binding.issueId }); } } @@ -9474,7 +9475,8 @@ async function executePaperclipNativeSessionWithinScope( warmNativeSessions.set(warmSessionId, { onReleased: () => releaseRetainedNativeSessionCleanup(input.db, { companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, - }), + }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId, + issueId: input.execution.binding.issueId }), pluginTaskId: usesPluginTask && input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "sandbox" ? input.runnerExecutionTarget.sandboxLeaseAcquisition?.providerLeaseId : undefined, runId: input.execution.binding.runId, @@ -11815,7 +11817,8 @@ export async function createRunnerdBackend(input: { initializingSessionToolAuthorities.delete(sessionScopeId); releaseRetainedNativeSessionCleanup(input.db, { companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, - }); + }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId, + issueId: input.execution.binding.issueId }); } } @@ -12860,7 +12863,7 @@ async function createRunnerdBackendWithinSessionClaim( }) .where(eq(environmentLeases.id, leaseId)) .returning({ id: environmentLeases.id }) - .then((rows) => rows[0] ?? null), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined }); + .then((rows) => rows[0] ?? null), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined, leaseId }); if (!updated) throw new Error("runner_harness_backup_lease_missing"); }; diff --git a/server/src/services/retained-native-session-cleanup-notifications.ts b/server/src/services/retained-native-session-cleanup-notifications.ts index d5aaff7c31a..e29665b3171 100644 --- a/server/src/services/retained-native-session-cleanup-notifications.ts +++ b/server/src/services/retained-native-session-cleanup-notifications.ts @@ -10,15 +10,15 @@ export type RetainedNativeSessionCleanupScope = { }; export async function notifyRetainedNativeSessionCleanup(tx: object, scope: RetainedNativeSessionCleanupScope) { - await notifyNativeHarnessBackupExpiration(tx, { companyId: scope.companyId }); + await notifyNativeHarnessBackupExpiration(tx, scope); await notifySkillVersionRetention(tx, scope.companyId); await notifyDeliveryWork(tx, DELIVERY_QUEUES.retainedNativeSessionCleanup, JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId, agentId: scope.agentId })); } -const listeners = new WeakMap void>>(); -export function onRetainedNativeSessionCleanupRelease(db: object, listener: (scope: RetainedNativeSessionCleanupScope) => void) { +const listeners = new WeakMap void>>(); +export function onRetainedNativeSessionCleanupRelease(db: object, listener: (scope: RetainedNativeSessionCleanupScope, source: RetainedNativeSessionCleanupScope) => void) { let owned = listeners.get(db); if (!owned) { owned = new Set(); listeners.set(db, owned); } owned.add(listener); @@ -27,6 +27,7 @@ export function onRetainedNativeSessionCleanupRelease(db: object, listener: (sco if (!owned.size && listeners.get(db) === owned) listeners.delete(db); }; } -export function releaseRetainedNativeSessionCleanup(db: object, scope: RetainedNativeSessionCleanupScope) { - for (const listener of listeners.get(db) ?? []) listener(scope); +export function releaseRetainedNativeSessionCleanup(db: object, scope: RetainedNativeSessionCleanupScope, + source: RetainedNativeSessionCleanupScope = scope) { + for (const listener of listeners.get(db) ?? []) listener(scope, source); }