From 75fa37f30fc1e65d1e936fcc58ad507cb53f88bb Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:31:31 -0500 Subject: [PATCH] fix(server): expire native backups from retention deadlines (#18380) Reviewed and verified at 5b137b5cdaf968b6545a7ac7a3137849240fd196. Co-Authored-By: Paperclip --- .../2026-10-30-backup-expiration-sleep.md | 63 +++ .../src/drivers/acpx/antigravity-auth.test.ts | 21 +- .../native-harness-backup-expiration.test.ts | 518 +++++++++++++++++- .../server-startup-feedback-export.test.ts | 10 +- server/src/index.ts | 36 +- .../services/delivery-work-notifications.ts | 1 + server/src/services/environments.ts | 6 +- server/src/services/issues.ts | 4 + ...ve-environment-settlement-notifications.ts | 4 +- ...harness-backup-expiration-notifications.ts | 26 + .../native-harness-backup-expiration-work.ts | 127 +++++ .../native-harness-backup-expiration.ts | 187 +++++-- .../native-harness-backup-publication.test.ts | 44 ++ .../native-harness-backup-publication.ts | 28 + .../native-session-executor.test.ts | 22 +- .../native-runtime/native-session-executor.ts | 51 +- ...ed-native-session-cleanup-notifications.ts | 11 +- 17 files changed, 1015 insertions(+), 144 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 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/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..307b4f67f47 --- /dev/null +++ b/doc/plans/2026-10-30-backup-expiration-sleep.md @@ -0,0 +1,63 @@ +# 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 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 +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. + +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 +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. +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. + +Review validation also covers expired-only directories whose listing fails with +EIO or EACCES. Only ENOENT means absent; other listing failures retain retry debt. +Both controls failed before the fix; all 104 expiration tests pass afterward. +Executor release tests assert both the preserved broad cleanup scope and the +precise source identity used for backup invalidation. 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/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index cd242bd6392..c5b72f9fc54 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -1,3 +1,12 @@ +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"; +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"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { registerNativeRunnerEvidenceRetentionWork } from "../services/native-runtime/native-runner-evidence-retention-work.js"; @@ -8,13 +17,15 @@ 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, companies, createDb, environmentLeases, + environments, + sandboxCreationOperations, heartbeatRuns, issues, nativeRunFinalizations, @@ -55,7 +66,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); @@ -75,6 +89,7 @@ describeEmbeddedPostgres("native harness backup expiration", () => { }); async function seedSession(input: { + companyId?: string; agentId?: string; issueStatus?: string; closedDaysAgo?: number; runStatus?: string; @@ -84,9 +99,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({ @@ -567,6 +583,481 @@ 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 }); + 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"))); + } + 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("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); + 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("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("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.each([false, true])("sandbox attachment reconciles only its retained root using durable run authority (explicit run: %s)", async explicitRun => { + const own = await seedSession({ runStatus: "running" }); + const unrelated = []; + for (let index = 0; index < 16; index++) unrelated.push(await seedSession({ companyId: own.companyId, + agentId: own.agentId, issueStatus: "todo", runStatus: index % 2 ? "running" : "succeeded" })); + const [environment] = await db.insert(environments).values({ name: randomUUID(), driver: "sandbox" }).returning(); + async function attach(runId: string) { + const operationId = randomUUID(); + await db.insert(sandboxCreationOperations).values({ id: operationId, companyId: own.companyId, + environmentId: environment!.id, originalEnvironmentId: environment!.id, runId, + provider: "controlled", driverKey: "controlled", config: {}, metadata: {}, nextAttemptAt: new Date() }); + return environmentService(db).acquireLease({ creationOperationId: operationId, companyId: own.companyId, + environmentId: environment!.id, ...(explicitRun ? { heartbeatRunId: runId } : {}) }); + } + await startBackups(); await settled(); + const reads = vi.spyOn(fs, "readFile"); + try { + const lease = await attach(own.runId); await settled(); + expect(reads.mock.calls.map(call => String(call[0]))) + .toEqual([path.join(own.root, "control-plane", "control-plane-state.json")]); + expect((await db.select().from(sandboxCreationOperations).where(eq(sandboxCreationOperations.leaseId, lease.id)))[0]) + .toMatchObject({ state: "attached", runId: own.runId }); + reads.mockClear(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: own.companyId, agentId: own.agentId, status: "running" }).returning(); + await attach(run!.id); await settled(); + expect(reads).not.toHaveBeenCalled(); + expect(await exists(own.current)).toBe(true); + for (const other of unrelated) expect(await exists(other.current)).toBe(true); + await quiet(); + } finally { + reads.mockRestore(); + await db.delete(sandboxCreationOperations).where(eq(sandboxCreationOperations.companyId, own.companyId)); + await db.delete(environmentLeases).where(eq(environmentLeases.environmentId, environment!.id)); + await db.delete(environments).where(eq(environments.id, environment!.id)); + } + }); + + 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.each(["EIO", "EACCES"])("retains expired-only cleanup after a directory read failure (%s)", async code => { + const session = await seedSession(); + const backupRoot = path.dirname(session.current); + const expired = path.join(backupRoot, `.expired-${randomUUID()}`); + await fs.rename(session.current, expired); + const read = fs.readdir; + const fault = vi.spyOn(fs, "readdir").mockImplementation(async (...args) => { + if (String(args[0]) === backupRoot) throw Object.assign(new Error("controlled cleanup listing failure"), { code }); + return read(...args); + }); + await startBackups(); + expect(backupErrors).toHaveLength(1); + const retry = coordinator!.nextWakeAt(); + expect(retry).not.toBeNull(); + expect(retry! - Date.now()).toBeGreaterThanOrEqual(29_000); + expect(await exists(session.current)).toBe(false); + expect(await exists(expired)).toBe(true); + fault.mockRestore(); + await quiet(5000); + await vi.advanceTimersByTimeAsync(retry! - Date.now()); await settled(); + expect(await fs.readdir(backupRoot)).toEqual([]); + 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) => { + 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" }); + 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"); @@ -608,8 +1099,12 @@ describeEmbeddedPostgres("native harness backup expiration", () => { it("retries a busy evidence obligation and disarms after the session releases it", async () => { const session = await seedSession({ issueStatus: "in_progress" }); const archive = await seedArchive(session.root); let busy = true; - await startEvidence({ isSessionBusy: () => busy }); - const retryAt = coordinator!.nextWakeAt()!; expect(retryAt - Date.now()).toBeGreaterThanOrEqual(60 * 60_000); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); vi.setSystemTime(now); + let inspectedAt: number | undefined; + await startEvidence({ isSessionBusy: () => { inspectedAt = Date.now(); return busy; } }); + await vi.advanceTimersByTimeAsync(50); + const retryAt = coordinator!.nextWakeAt()!; expect(inspectedAt).toBeDefined(); + expect(retryAt - inspectedAt!).toBeGreaterThanOrEqual(60 * 60_000); await quiet(5000); busy = false; await vi.advanceTimersByTimeAsync(retryAt - Date.now()); await settled(); expect(await exists(archive)).toBe(false); await quiet(); @@ -637,11 +1132,18 @@ describeEmbeddedPostgres("native harness backup expiration", () => { it("backs off a failed state-base inspection and lets publication reactivate it", async () => { const session = await seedSession({ issueStatus: "in_progress" }); const archive = await seedArchive(session.root), read = fs.readdir; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); vi.setSystemTime(now); + let inspectedAt: number | undefined; const denied = vi.spyOn(fs, "readdir").mockImplementation(async (...args) => { - if (args[0] === stateBase) throw Object.assign(new Error("state base unavailable"), { code: "EACCES" }); + if (args[0] === stateBase) { + inspectedAt = Date.now(); + throw Object.assign(new Error("state base unavailable"), { code: "EACCES" }); + } return read(...args); }); - await startEvidence(); expect(coordinator!.nextWakeAt()! - Date.now()).toBeGreaterThanOrEqual(60 * 60_000); + await startEvidence(); await vi.advanceTimersByTimeAsync(50); + expect(inspectedAt).toBeDefined(); + expect(coordinator!.nextWakeAt()! - inspectedAt!).toBeGreaterThanOrEqual(60 * 60_000); denied.mockRestore(); await quiet(5 * 60_000); expect(await exists(archive)).toBe(true); noteNativeRunnerEvidence(now.getTime() - 10 * DAY_MS); await settled(); expect(await exists(archive)).toBe(false); await quiet(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index cea50f28f75..63b607e0019 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -592,6 +592,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(), })), })); @@ -767,9 +769,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 () => { @@ -777,6 +782,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/index.ts b/server/src/index.ts index a378b376020..21fe920a6ce 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 { registerMergeConfirmationWork } from "./services/merged-pr-confirmation-work.js"; import { registerRoutineExecutionStatusWork } from "./services/routine-execution-status-work.js"; import { routineExecutionStatusService } from "./services/routine-run-status.js"; @@ -36,8 +37,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"; @@ -1293,30 +1292,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, @@ -1351,6 +1326,12 @@ async function startServerWithDatabaseTeardown( () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby()); 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; @@ -2093,9 +2074,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 b3f96f0170c..bd7bd93c8bc 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -134,6 +134,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/environments.ts b/server/src/services/environments.ts index e46df62ed7d..68fe46ede14 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -68,7 +68,7 @@ export async function attachSandboxCreation(db: Db, id: string, scope: { company } if (!operation || operation.state !== "pending") throw new Error("Sandbox creation ownership has already transferred"); const lease = await attach(tx as unknown as Db); - await notifyEnvironmentCleanup(tx, { companyId: scope.companyId }); + await notifyEnvironmentCleanup(tx, { companyId: scope.companyId, runId: operation.runId }); await tx.update(sandboxCreationOperations).set({ state: "attached", leaseId: lease.id, updatedAt: new Date() }) .where(eq(sandboxCreationOperations.id, id)); return lease; @@ -1705,7 +1705,7 @@ export function environmentService(db: Db) { creationOperationId: sql`${environmentLeases.metadata}->>'creationOperationId'`, }).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))) { @@ -1830,7 +1830,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 f3e35046ac9..2455b28ef99 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 { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js"; import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; @@ -11540,6 +11541,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, issueId: id }); + } await notifyStatusCardIssueChange(tx, receiptExisting.companyId); 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 5769b81015a..0c55810b6a8 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 type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string; leaseId?: string }; export async function notifyNativeEnvironmentSettlement(tx: object, scope?: NativeEnvironmentSettlementScope) { + 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 new file mode 100644 index 00000000000..e334487a82c --- /dev/null +++ b/server/src/services/native-harness-backup-expiration-notifications.ts @@ -0,0 +1,26 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +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 ? nativeHarnessBackupScopeHint(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..174f804b3a5 --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-expiration-work.ts @@ -0,0 +1,127 @@ +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 { 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 NativeHarnessBackupDependencies, type NativeHarnessBackupExpirationOptions } from "./native-harness-backup-expiration.js"; + +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, + options: NativeHarnessBackupExpirationOptions, canRun: () => boolean, +) { + const service = nativeHarnessBackupExpirationService(db, { ...options, duty: "backups" }); + const cursors = new Map(); + 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); } + } + } + async function reconcile(id: string): Promise { + const source = await service.reconcileBackup(id); + 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, + load: async (scope, signal) => { + const hint = scope ? nativeHarnessBackupScopeHint(scope) : "*"; + if (scope && !cursors.has(hint)) await discoverReferences(scope); + const limit = options.maxEntries ?? 100; + 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(hint); else cursors.set(hint, 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 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 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(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 34402b8ad2d..6e5ab7c424d 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,17 +163,22 @@ async function readSessionIdentity(root: string): Promise[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[] }> { + 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 }) .from(heartbeatRuns) @@ -200,7 +208,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, @@ -208,8 +216,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" }; @@ -249,7 +257,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: 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) { @@ -347,7 +355,7 @@ export function nativeHarnessBackupExpirationService( async function settleLeftovers(backupRoot: string) { const current = path.join(backupRoot, "current"); - for (const name of await fs.readdir(backupRoot).catch(() => [] as string[])) { + for (const name of (await fs.readdir(backupRoot).catch(missingPath)) ?? []) { const target = path.join(backupRoot, name); if (name.startsWith(RETIRING_PREFIX) && UUID_PATTERN.test(name.slice(RETIRING_PREFIX.length))) { if (!(await isRealDirectory(current))) await fs.rename(target, current); @@ -382,54 +390,49 @@ 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; + 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 }); + 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) { @@ -748,5 +751,91 @@ export function nativeHarnessBackupExpirationService( }); } - return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration }; + 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); + if (!SCOPE_PATTERN.test(scopeDigest) || !(await isRealDirectory(root))) return null; + const backupRoot = path.join(root, "failover-backups"); + if (!(await isRealDirectory(backupRoot))) return null; + if (!defaults.dryRun) await settleLeftovers(backupRoot); + const current = path.join(backupRoot, "current"); + const stat = await fs.lstat(current).catch(missingPath); + if (!stat?.isDirectory() || stat.isSymbolicLink()) return null; + const identity = await readSessionIdentity(root); + if (!identity) return null; + 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), 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 { ...source, dueAt: Math.max(stat!.mtime.getTime(), 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) { + 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; + 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, + reconcileBackup, backupScopes, expireBackup, resolveBackupScope }; } 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.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index dc82054a0eb..ae3c6a1a493 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -5304,7 +5304,10 @@ describe("retained native cleanup activation", () => { } const outcome = await pendingOutcome; if (mode === "legacy_busy_copy") { - expect(maintenanceReleased).toHaveBeenCalledExactlyOnceWith({ companyId: run.companyId, runId: run.id }); + expect(maintenanceReleased).toHaveBeenCalledExactlyOnceWith( + { companyId: run.companyId, runId: run.id }, + { companyId: run.companyId, runId: run.id }, + ); unsubscribeMaintenance?.(); } if (canonicalSource) { @@ -6626,7 +6629,10 @@ describe("native startup cancellation fence", () => { })).rejects.toThrow("native_cancellation_pending_recovery"); expect(cancel).toHaveBeenCalledOnce(); expect(submit).not.toHaveBeenCalled(); - expect(released).toHaveBeenCalledWith({ companyId: execution.binding.companyId, agentId: execution.binding.agentId }); + expect(released).toHaveBeenCalledWith( + { companyId: execution.binding.companyId, agentId: execution.binding.agentId }, + { companyId: execution.binding.companyId, runId: execution.binding.runId, issueId: execution.binding.issueId }, + ); } finally { unsubscribe(); } }); }); @@ -10264,9 +10270,10 @@ describe("native warm session supervision", () => { ); expect(released).not.toHaveBeenCalled(); releaseClose(); - await vi.waitFor(() => expect(released).toHaveBeenCalledExactlyOnceWith({ - companyId: base.binding.companyId, agentId: base.binding.agentId, - })); + await vi.waitFor(() => expect(released).toHaveBeenCalledExactlyOnceWith( + { companyId: base.binding.companyId, agentId: base.binding.agentId }, + { companyId: base.binding.companyId, runId: base.binding.runId, issueId: base.binding.issueId }, + )); } finally { releaseClose(); unsubscribe(); } }); @@ -12915,7 +12922,10 @@ describe("runnerd provider runtime wiring", () => { releaseEnvironment(); releaseAssigned(); const result = await preparation; - expect(released).toHaveBeenCalledExactlyOnceWith({ companyId: execution.binding.companyId, agentId: execution.binding.agentId }); + expect(released).toHaveBeenCalledExactlyOnceWith( + { companyId: execution.binding.companyId, agentId: execution.binding.agentId }, + { companyId: execution.binding.companyId, runId: execution.binding.runId, issueId: execution.binding.issueId }, + ); if (failure !== "success") { expect(result).toMatchObject({ error: { message: failure === "assigned" ? "assigned failed" : "environment failed" } }); expect(state.createBackend).not.toHaveBeenCalled(); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c78016c1ad6..c12cbc1bf52 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 { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; import { rehydratePinnedSkillAssets } from "./runtime-context.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeRecoveryUsageDelivery } from "./native-recovery-usage-delivery.js"; @@ -8250,7 +8251,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 }); } } @@ -9693,7 +9695,8 @@ async function executePaperclipNativeSessionWithinScope( sessionCheckpointPersisted: runnerdBackend?.sessionCheckpointPersisted, 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, @@ -12254,7 +12257,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 }); } } @@ -13212,7 +13216,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"); }; @@ -13644,40 +13648,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); + } : undefined, + }); } 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..e29665b3171 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,14 +10,15 @@ export type RetainedNativeSessionCleanupScope = { }; export async function notifyRetainedNativeSessionCleanup(tx: object, scope: RetainedNativeSessionCleanupScope) { + 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); @@ -25,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); }