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 1/2] 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); } From f77620b9a4b76cc05508612ee894a20b751651bb Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 14:50:21 -0500 Subject: [PATCH 2/2] Refresh GitHub event connector after live enrollment Resolve the current identity only on admitted work, retain unchanged connector negotiation, and replace removed or rotated identity configuration. Prove real signed enrollment and encrypted event delivery without recreating the registered worker or adding idle polling. Co-Authored-By: Paperclip --- .../2026-10-30-github-connection-sleep.md | 25 ++++ .../github-connection-enrollment-work.test.ts | 139 ++++++++++++++++++ .../src/services/github-connection-events.ts | 19 ++- 3 files changed, 181 insertions(+), 2 deletions(-) create mode 100644 server/src/__tests__/github-connection-enrollment-work.test.ts diff --git a/doc/plans/2026-10-30-github-connection-sleep.md b/doc/plans/2026-10-30-github-connection-sleep.md index 3e6de1148bb..287b479b07c 100644 --- a/doc/plans/2026-10-30-github-connection-sleep.md +++ b/doc/plans/2026-10-30-github-connection-sleep.md @@ -114,3 +114,28 @@ concurrent replay, signed callbacks, outer commit/rollback, drain/resume, startu wiring in both scheduler branches, credential backoff and changed deadlines. Cloud acceptance uses its actual broker/store/notifier boundaries and a controlled HTTP tenant. These tests do not establish a deployed fleet rollout or cost savings. + +## Enrollment after worker startup + +The event owner resolves the current connector only when a delivery or setup pass +is admitted. A service created before self-host enrollment can therefore use a +later completed enrollment without being recreated. The full normalized identity +and target configuration keys the connector cache, preserving negotiated event +wake support while it is unchanged. An absent identity clears the cache; a +replacement identity or changed key creates a new connector. Explicitly injected +connectors keep their original lifetime. + +Enrollment does not add a timer, database scan or network probe. An ordinary +credential/configuration producer, authenticated control wake or active setup +request admits the next pass. Unconfigured passes return before owner SQL or +HTTP. A pass already in flight keeps its connector for that lease and its matching +acknowledgment; identity discovery never changes authority halfway through a pass. + +The real PostgreSQL/loopback regression starts the registered owner unenrolled, +completes the canonical signed enrollment flow, then applies and acknowledges an +encrypted installation event. Removing the identity parks the same owner, and a +second enrollment proves the replacement signing/sealing identity is used. Each +applied installation update has one bounded empty follow-up from its credential +hint; the test joins that saved deadline once before checking 25-minute SQL/HTTP +quiet windows. This is process-local enrollment recovery, not a deployed Cloud +rollout or a new cancellation policy. diff --git a/server/src/__tests__/github-connection-enrollment-work.test.ts b/server/src/__tests__/github-connection-enrollment-work.test.ts new file mode 100644 index 00000000000..c888565b3d9 --- /dev/null +++ b/server/src/__tests__/github-connection-enrollment-work.test.ts @@ -0,0 +1,139 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createServer, type Server } from "node:http"; +import { once } from "node:events"; +import { createCipheriv, createPublicKey, diffieHellman, generateKeyPairSync, hkdfSync, randomBytes, randomUUID, verify } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { companies, connectionEventDeliveries, connectionGrants, createDb, toolApplications, toolConnections } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { githubConnectionEventService } from "../services/github-connection-events.js"; +import { registerGitHubConnectionEventWork } from "../services/github-connection-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { wakeConnectionEvents } from "../services/connection-event-wake.js"; +import { completePaperclipCloudConnectorEnrollment, loadPaperclipCloudConnectorIdentity, startPaperclipCloudConnectorEnrollment } from "../services/paperclip-cloud-connector-enrollment.js"; +import { paperclipCloudConnectorIdentityPath } from "../services/paperclip-cloud-connector-enrollment.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("GitHub enrollment on a live event worker (real PostgreSQL and signed loopback Cloud)", () => { + let temporary: Awaited>; + let db: ReturnType; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("github-live-enrollment-"); + db = createDb(temporary.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 90_000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + + it("enrolls without recreation, drains an authenticated event, forgets removed identity and re-enrolls from quiet", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "github-live-enrollment-")); + const previous = { home: process.env.PAPERCLIP_HOME, instance: process.env.PAPERCLIP_INSTANCE_ID }; + process.env.PAPERCLIP_HOME = root; process.env.PAPERCLIP_INSTANCE_ID = "enrollment-worker"; + let server: Server | undefined; + let worker: ReturnType | undefined; + let coordinator: ReturnType | undefined; + const errors: unknown[] = [], httpErrors: unknown[] = []; + let requests = 0, acknowledged = 0, leases = 0, enrollment: Record | undefined; + let origin = "", eventId = randomUUID(), grantId = "", companyId = ""; + const identities: string[] = []; + const delivered = new Set(); + try { + server = createServer(async (req, res) => { + try { + requests++; + const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(Buffer.from(chunk)); + const body = JSON.parse(Buffer.concat(chunks).toString()); + if (req.url === "/v1/connector/enrollments") { + enrollment = body; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ enrollmentId: "enrollment", verificationUrl: `${origin}/connections/enroll?id=enrollment`, expiresAt: new Date(Date.now() + 60_000).toISOString() })); return; + } + expect(enrollment).toBeDefined(); + const [header, payload, signature] = body.request.split("."); + expect(verify(null, Buffer.from(`${header}.${payload}`), createPublicKey({ key: { kty: "OKP", crv: "Ed25519", x: enrollment!.signPublicKey }, format: "jwk" }), Buffer.from(signature, "base64url"))).toBe(true); + const claims = JSON.parse(Buffer.from(payload, "base64url").toString()); + expect(claims.iss).toBe(enrollment!.instanceId); expect(claims.env).toBe("test"); expect(claims.aud).toBe(`${origin}${req.url}`); + res.setHeader("content-type", "application/json"); + if (req.url === "/v1/connector/enrollment-claims") { + res.end(JSON.stringify({ id: enrollment!.instanceId, environment: "test", origins: [enrollment!.origin] })); return; + } + identities.push(claims.iss); + if (req.url === "/v1/connector/events/ack") { acknowledged++; delivered.add(eventId); res.end(JSON.stringify({ acknowledged: 1 })); return; } + expect(req.url).toBe("/v1/connector/events/lease"); leases++; + if (delivered.has(eventId)) { res.end(JSON.stringify({ events: [], eventWakeVersion: 1 })); return; } + const ephemeral = generateKeyPairSync("x25519"), epk = ephemeral.publicKey.export({ format: "jwk" }).x!; + const recipient = createPublicKey({ key: { kty: "OKP", crv: "X25519", x: enrollment!.sealPublicKey }, format: "jwk" }); + const aad = Buffer.from([1, "X25519-HKDF-SHA256-A256GCM", "events", claims.iss, "test", "github", "github.code", ""].join("\n")); + const key = Buffer.from(hkdfSync("sha256", diffieHellman({ privateKey: ephemeral.privateKey, publicKey: recipient }), Buffer.concat([Buffer.from(epk, "base64url"), Buffer.from(enrollment!.sealPublicKey, "base64url")]), aad, 32)); + const iv = randomBytes(12), cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad); + const packet = { v: 1, instanceId: claims.iss, environment: "test", leaseId: "lease", events: [{ id: eventId, provider: "github", event: "installation", action: "unsuspend", installationId: "101", repositoryId: null, createdAt: new Date().toISOString(), bindingIds: [`${grantId}_101`], payload: { installationId: "101" } }] }; + const ct = Buffer.concat([cipher.update(JSON.stringify(packet)), cipher.final(), cipher.getAuthTag()]); + res.end(JSON.stringify({ leaseId: "lease", eventWakeVersion: 1, sealed: { v: 1, alg: "X25519-HKDF-SHA256-A256GCM", purpose: "events", provider: "github", profile: "github.code", epk, iv: iv.toString("base64url"), ct: ct.toString("base64url") } })); + } catch (error) { httpErrors.push(error); res.writeHead(500).end(); } + }); + const listening = once(server, "listening"); server.listen(0, "127.0.0.1"); await listening; + const address = server.address(); if (!address || typeof address === "string") throw new Error("No loopback listener"); + origin = `http://127.0.0.1:${address.port}`; + companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Live enrollment", issuePrefix: "GLE" }); + const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: randomUUID(), name: "GitHub", type: "mcp_http", status: "active" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app!.id, uid: randomUUID(), name: "GitHub", status: "active", enabled: true, transport: "mcp_remote", authKind: "oauth", credentialPolicy: "per_user", config: { sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } }, transportConfig: {} }).returning(); + const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId: connection!.id, kind: "user", subjectUserId: "owner", status: "active", providerTenant: { github: { userId: "42", login: "octocat", installationIds: ["101"], installationCount: 1, repositoryCount: 0, repositorySelection: "all", installationOwnerLogins: [], webhookHealth: "unhealthy" } } }).returning(); + grantId = grant!.id; + const env = { PAPERCLIP_CLOUD_CONNECTOR_BASE_URL: origin, PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT: "test" }; + const service = githubConnectionEventService(db, { env }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: e => errors.push(e) }); + const selects = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); + worker = registerGitHubConnectionEventWork(db, coordinator, service, () => true); + await worker.ready; + async function quiet(stage: string) { + await vi.waitFor(() => { expect(idleWorkSnapshot().active, stage).toBe(0); expect(coordinator!.nextWakeAt(), stage).toBeNull(); }); + selects.mockClear(); transactions.mockClear(); executes.mockClear(); const before = requests; + await vi.advanceTimersByTimeAsync(25 * 60_000); + expect(selects).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(requests).toBe(before); + } + await quiet("unenrolled startup"); expect(requests).toBe(0); + async function enroll() { + await startPaperclipCloudConnectorEnrollment({ origin: "https://self-hosted.example.test", env, companyId }); + const pending = loadPaperclipCloudConnectorIdentity()!; + await completePaperclipCloudConnectorEnrollment({ enrollmentId: pending.pending!.enrollmentId, state: pending.pending!.returnState, approvalCode: "approved", env }); + return loadPaperclipCloudConnectorIdentity()!.instanceId; + } + const firstIdentity = await enroll(); + expect(wakeConnectionEvents(db)).toBe(true); + await vi.advanceTimersByTimeAsync(6_000); + await vi.waitFor(() => expect(acknowledged).toBe(1)); + expect(await db.select().from(connectionEventDeliveries).where(eq(connectionEventDeliveries.providerDeliveryId, eventId))).toMatchObject([{ status: "processed" }]); + expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grantId)))[0]!.providerTenant!.github!.webhookHealth).toBe("healthy"); + // The applied installation update publishes one credential-change hint. + // Join that existing minimum-spacing pass once; the broker now has no debt. + const followup = coordinator.nextWakeAt(); expect(followup).not.toBeNull(); + expect(followup! - Date.now()).toBeLessThanOrEqual(5_000); + await vi.advanceTimersByTimeAsync(Math.max(0, followup! - Date.now()) + 1); + await vi.waitFor(() => expect(leases).toBe(2)); + await quiet("first event plus its one empty follow-up"); + await rm(paperclipCloudConnectorIdentityPath()); + const oldLeases = leases; expect(wakeConnectionEvents(db)).toBe(true); await vi.advanceTimersByTimeAsync(6_000); + await quiet("removed identity"); expect(leases).toBe(oldLeases); + eventId = randomUUID(); const secondIdentity = await enroll(); expect(secondIdentity).not.toBe(firstIdentity); + expect(wakeConnectionEvents(db)).toBe(true); await vi.advanceTimersByTimeAsync(6_000); + await vi.waitFor(() => expect(acknowledged).toBe(2)); + expect(identities).toContain(firstIdentity); expect(identities).toContain(secondIdentity); + expect(identities.slice(identities.indexOf(secondIdentity))).toEqual(expect.arrayContaining([secondIdentity, secondIdentity])); + expect(identities.slice(identities.indexOf(secondIdentity)).every(id => id === secondIdentity)).toBe(true); + const replacementFollowup = coordinator.nextWakeAt(); expect(replacementFollowup).not.toBeNull(); + expect(replacementFollowup! - Date.now()).toBeLessThanOrEqual(5_000); + await vi.advanceTimersByTimeAsync(Math.max(0, replacementFollowup! - Date.now()) + 1); + await vi.waitFor(() => expect(leases).toBe(oldLeases + 2)); + await quiet("replacement event plus its one empty follow-up"); expect(errors).toEqual([]); expect(httpErrors).toEqual([]); + } finally { + await worker?.stop(); await coordinator?.stop(); vi.restoreAllMocks(); vi.useRealTimers(); + if (server) { const closed = once(server, "close"); server.close(); server.closeAllConnections(); await closed; } + if (previous.home === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previous.home; + if (previous.instance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = previous.instance; + await rm(root, { recursive: true, force: true }); + } + }, 30_000); +}); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index a70213dc44b..af936353bdd 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -4,6 +4,7 @@ import { notifyToolConnectionHealth } from "./tool-connection-health-notificatio import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js"; import { chatEndpoints } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; import { connectionEventDeliveries, connectionGrants, @@ -223,8 +224,21 @@ export function githubConnectionEventService( ) { const now = options.now ?? (() => new Date()); let nextPollAt = 0; - const config = options.connector ? null : paperclipCloudConnectorConfigFromEnv(options.env); - const connector = options.connector ?? (config ? createPaperclipCloudConnector({ config }) : null); + let configuredConnector: { + config: NonNullable>; + connector: PaperclipCloudConnector; + } | undefined; + function currentConnector() { + if (options.connector) return options.connector; + // Enrollment and identity replacement can happen after this owner starts. + // Read only on an admitted pass; preserve negotiated delivery while unchanged. + const config = paperclipCloudConnectorConfigFromEnv(options.env); + if (!config) { configuredConnector = undefined; return null; } + if (!configuredConnector || !isDeepStrictEqual(configuredConnector.config, config)) { + configuredConnector = { config, connector: createPaperclipCloudConnector({ config }) }; + } + return configuredConnector.connector; + } const forcePolling = (options.env ?? process.env).PAPERCLIP_GITHUB_EVENT_POLLING === "1"; let emptyPolls = 0; let nextSetupPollAt = 0; @@ -484,6 +498,7 @@ export function githubConnectionEventService( } async function runPoll(): Promise { + const connector = currentConnector(); if (!connector) { nextPollAt = Number.POSITIVE_INFINITY; return { leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 };