mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 16:58:52 +01:00
Integrate scoped backup expiration notifications
Preserve backup issue identity alongside existing recovery, retention, card, routine, merge, and external-object owners. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
commit
04afdcdee5
16 files changed
+913
-132
No files matched your search
@@ -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,7 +17,7 @@ import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promises as fs } from "node:fs";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import {
|
||||
agents,
|
||||
agentTaskSessions,
|
||||
@@ -55,7 +64,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let worker: ReturnType<typeof registerNativeRunnerEvidenceRetentionWork> | undefined;
|
||||
let admitted = true;
|
||||
let backupWorker: ReturnType<typeof registerNativeHarnessBackupExpirationWork> | 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 +87,7 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
|
||||
});
|
||||
|
||||
async function seedSession(input: {
|
||||
companyId?: string; agentId?: string;
|
||||
issueStatus?: string;
|
||||
closedDaysAgo?: number;
|
||||
runStatus?: string;
|
||||
@@ -84,9 +97,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
|
||||
cleanupActive?: boolean;
|
||||
} = {}) {
|
||||
if (!stateBase) stateBase = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-durable-sessions-"));
|
||||
const companyId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Backups", issuePrefix: `B${companyId.slice(0, 6)}` });
|
||||
const [agent] = await db.insert(agents).values({ companyId, name: "Coder" }).returning();
|
||||
const companyId = input.companyId ?? randomUUID();
|
||||
if (!input.companyId) await db.insert(companies).values({ id: companyId, name: "Backups", issuePrefix: `B${companyId.slice(0, 6)}` });
|
||||
const [agent] = input.agentId ? [{ id: input.agentId }]
|
||||
: await db.insert(agents).values({ companyId, name: "Coder" }).returning();
|
||||
const issueStatus = input.issueStatus ?? "done";
|
||||
const closedAt = new Date(now.getTime() - (input.closedDaysAgo ?? 10) * DAY_MS);
|
||||
const [issue] = await db.insert(issues).values({
|
||||
@@ -558,6 +572,420 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
|
||||
reads.mockRestore(); writes.mockRestore(); directories.mockRestore();
|
||||
}
|
||||
|
||||
async function startBackups(options: Parameters<typeof nativeHarnessBackupExpirationService>[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<void>(resolve => { release = resolve; });
|
||||
const staged = new Promise<void>(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("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<void>(), release = Promise.withResolvers<void>();
|
||||
const held = vi.spyOn(fs, "lstat").mockImplementation(async (...args) => {
|
||||
if (String(args[0]) === path.join(own.current, "codex-home", "history.jsonl")) {
|
||||
entered.resolve(); await release.promise;
|
||||
}
|
||||
return lstat(...args);
|
||||
});
|
||||
const startup = startBackups(); await entered.promise;
|
||||
try {
|
||||
await db.transaction(async tx => {
|
||||
await tx.execute(sql`set local lock_timeout = '500ms'`);
|
||||
await issueService(db).update(own.issueId, { status: "todo" }, tx);
|
||||
});
|
||||
} finally { release.resolve(); await startup; held.mockRestore(); }
|
||||
await settled(); expect(await exists(own.current)).toBe(true); expect(backupErrors).toEqual([]); await quiet();
|
||||
});
|
||||
|
||||
it.each(["EIO", "EACCES"])("retains a failed identity read as real retry debt (%s)", async code => {
|
||||
const own = await seedSession(); const read = fs.readFile;
|
||||
const fault = vi.spyOn(fs, "readFile").mockImplementation(async (...args) => {
|
||||
if (String(args[0]) === path.join(own.root, "control-plane", "control-plane-state.json")) {
|
||||
throw Object.assign(new Error("controlled identity I/O failure"), { code });
|
||||
}
|
||||
return read(...args);
|
||||
});
|
||||
await startBackups(); expect(backupErrors).toHaveLength(1);
|
||||
const retry = coordinator!.nextWakeAt()!; expect(retry - Date.now()).toBeGreaterThanOrEqual(29_000);
|
||||
expect(await exists(own.current)).toBe(true); fault.mockRestore(); await quiet(5000);
|
||||
await vi.advanceTimersByTimeAsync(retry - Date.now()); await settled();
|
||||
expect(await exists(own.current)).toBe(false); expect(await taskSessionCount(own.agentId)).toBe(0); await quiet();
|
||||
});
|
||||
|
||||
it("reconstructs future retention without issue row locks", async () => {
|
||||
await seedSession({ closedDaysAgo: 2 }); const queries: string[] = [];
|
||||
const previous = db.$client.options.debug;
|
||||
db.$client.options.debug = (_connection, query) => { queries.push(query); };
|
||||
try {
|
||||
await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1);
|
||||
expect(queries.some(query => /for\s+share/i.test(query))).toBe(false);
|
||||
} finally { db.$client.options.debug = previous; }
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it("rejects destructive backup scopes outside the controller digest namespace", async () => {
|
||||
await emptyStateBase(); const remove = vi.spyOn(fs, "rm");
|
||||
await expect(service().expireBackup("../outside")).rejects.toThrow("Invalid native harness backup scope");
|
||||
expect(remove).not.toHaveBeenCalled(); remove.mockRestore();
|
||||
});
|
||||
|
||||
it.each([false, true])("finishes cleanup journals after ordinary agent deletion (restart %s)", async restart => {
|
||||
const session = await seedSession(); const remove = fs.rm;
|
||||
const fault = vi.spyOn(fs, "rm").mockImplementation(async (...args) => {
|
||||
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");
|
||||
|
||||
@@ -590,6 +590,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(),
|
||||
})),
|
||||
}));
|
||||
|
||||
@@ -765,9 +767,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 () => {
|
||||
@@ -775,6 +780,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 {
|
||||
|
||||
+7
-29
@@ -2,6 +2,7 @@ import { attentionRetentionService, registerAttentionRetentionWork } from "./ser
|
||||
import { registerRoutineExecutionStatusWork } from "./services/routine-execution-status-work.js";
|
||||
import { routineExecutionStatusService } from "./services/routine-run-status.js";
|
||||
import { registerMergeConfirmationWork } from "./services/merged-pr-confirmation-work.js";
|
||||
import { registerNativeHarnessBackupExpirationWork } from "./services/native-runtime/native-harness-backup-expiration-work.js";
|
||||
import { registerProviderTraceRetentionWork } from "./services/provider-trace-retention-work.js";
|
||||
import { registerRunEvidenceMigrationWork } from "./services/run-evidence-migration-work.js";
|
||||
import { registerSkillVersionRetentionWork } from "./services/skill-version-retention-work.js";
|
||||
@@ -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";
|
||||
@@ -1297,30 +1296,6 @@ async function startServerWithDatabaseTeardown(
|
||||
evidenceRetentionDays: config.nativeRunnerEvidenceRetentionDays,
|
||||
shouldContinue: () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby(),
|
||||
});
|
||||
let nativeHarnessBackupExpirationInFlight = false;
|
||||
let nativeHarnessBackupExpirationContinuation: ReturnType<typeof setTimeout> | 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,
|
||||
@@ -1355,6 +1330,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;
|
||||
@@ -2167,9 +2148,6 @@ async function startServerWithDatabaseTeardown(
|
||||
heartbeatSchedulerStopped = true;
|
||||
skillVersionRetentionWorker?.unsubscribe();
|
||||
await taskDrainExpiryWork.stop();
|
||||
clearInterval(nativeHarnessBackupExpirationInterval);
|
||||
clearTimeout(nativeHarnessBackupExpirationStartup);
|
||||
clearTimeout(nativeHarnessBackupExpirationContinuation);
|
||||
nativeRunnerEvidenceRetentionWorker?.unsubscribe();
|
||||
await skillSourceSync.stop();
|
||||
unsubscribeChatCompletions();
|
||||
|
||||
@@ -125,6 +125,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];
|
||||
|
||||
@@ -1705,7 +1705,7 @@ export function environmentService(db: Db) {
|
||||
creationOperationId: sql<string | null>`${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)),
|
||||
|
||||
@@ -4,6 +4,7 @@ import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"
|
||||
import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
|
||||
import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js";
|
||||
import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
|
||||
import { notifyNativeHarnessBackupExpiration } from "./native-harness-backup-expiration-notifications.js";
|
||||
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
|
||||
import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
|
||||
import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
|
||||
@@ -11507,6 +11508,9 @@ export function issueService(db: Db) {
|
||||
patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number;
|
||||
}
|
||||
await notifyStatusCardIssueChange(tx, receiptExisting.companyId);
|
||||
if (patch.status !== undefined || patch.completedAt !== undefined || patch.cancelledAt !== undefined) {
|
||||
await notifyNativeHarnessBackupExpiration(tx, { companyId: receiptExisting.companyId, issueId: id });
|
||||
}
|
||||
if ((patch.status !== undefined && patch.status !== receiptExisting.status)
|
||||
|| patch.completedAt !== undefined || patch.cancelledAt !== undefined
|
||||
|| issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) {
|
||||
|
||||
@@ -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 }) : "*");
|
||||
|
||||
@@ -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<object, Set<(scope: NativeHarnessBackupExpirationScope) => 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);
|
||||
}
|
||||
@@ -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<string>; issueIds: Set<string>; leaseIds: Set<string> };
|
||||
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<typeof createDeliveryWorkCoordinator>, db: Db,
|
||||
options: NativeHarnessBackupExpirationOptions, canRun: () => boolean,
|
||||
) {
|
||||
const service = nativeHarnessBackupExpirationService(db, { ...options, duty: "backups" });
|
||||
const cursors = new Map<string, string>();
|
||||
const sources = new Map<string, Dependencies>();
|
||||
const index = new Map<string, Set<string>>();
|
||||
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<string> {
|
||||
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<Entry | null> {
|
||||
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(); } };
|
||||
}
|
||||
@@ -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<SessionIdentity | null
|
||||
const stat = await fs.lstat(statePath);
|
||||
if (!stat.isFile() || stat.size > MAX_CONTROL_PLANE_STATE_BYTES) return null;
|
||||
const state = JSON.parse(await fs.readFile(statePath, "utf8")) as { identity?: Record<string, unknown> };
|
||||
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<SessionIdentity | null
|
||||
? environmentLeaseId
|
||||
: null,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
} catch (error) {
|
||||
if (error instanceof SyntaxError || ["ENOENT", "ENOTDIR"].includes((error as NodeJS.ErrnoException).code ?? "")) return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export type NativeHarnessBackupDependencies = {
|
||||
companyId: string; agentId: string; sessionId: string; runIds: string[]; issueIds: string[]; leaseIds: string[];
|
||||
};
|
||||
|
||||
type Tx = Parameters<Parameters<Db["transaction"]>[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)
|
||||
@@ -197,7 +205,7 @@ async function sessionRetentionBlocker(
|
||||
if (input.checkpoint && runs.some(run => run.finishedAt! >= input.closedBefore)) return { reason: "recent_run" };
|
||||
const issueIds = [...new Set(runs.map((run) => run.issueId).filter((id): id is string => Boolean(id)))];
|
||||
if (issueIds.length === 0) return { reason: "issue_unknown" };
|
||||
const issueRows = await tx
|
||||
const issueQuery = tx
|
||||
.select({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
@@ -205,8 +213,8 @@ async function sessionRetentionBlocker(
|
||||
closedAt: sql<Date | null>`coalesce(${issues.completedAt}, ${issues.cancelledAt})`,
|
||||
})
|
||||
.from(issues)
|
||||
.where(and(eq(issues.companyId, owner.companyId), inArray(issues.id, issueIds)))
|
||||
.for("share");
|
||||
.where(and(eq(issues.companyId, owner.companyId), inArray(issues.id, issueIds)));
|
||||
const issueRows = await (input.lock === false ? issueQuery : issueQuery.for("share"));
|
||||
if (issueRows.length !== issueIds.length) return { reason: "issue_unknown" };
|
||||
for (const issue of issueRows) {
|
||||
if (!CLOSED_ISSUE_STATUSES.includes(issue.status)) return { reason: "issue_open" };
|
||||
@@ -246,7 +254,7 @@ async function sessionRetentionBlocker(
|
||||
.where(and(inArray(environmentLeases.status, LIVE_LEASE_STATUSES), or(...leaseReferences)))
|
||||
.limit(1);
|
||||
if (lease) return { reason: "live_lease" };
|
||||
return { ...owner, runIds };
|
||||
return { ...owner, runIds, issueIds, closedAt: issueRows.reduce((latest, issue) => Math.max(latest, new Date(issue.closedAt!).getTime()), Number.NEGATIVE_INFINITY) };
|
||||
}
|
||||
|
||||
async function hasLiveRun(db: Db, owner: { companyId: string; agentId: string }, sessionId: string) {
|
||||
@@ -379,54 +387,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) {
|
||||
@@ -745,5 +748,91 @@ export function nativeHarnessBackupExpirationService(
|
||||
});
|
||||
}
|
||||
|
||||
return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration };
|
||||
async function backupDependencies(scopeDigest: string, identity: SessionIdentity): Promise<NativeHarnessBackupDependencies | null> {
|
||||
const [owner] = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
issueId: sql<string | null>`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<string | null>`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<string | null>`${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<string | null>`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 };
|
||||
}
|
||||
@@ -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 }]);
|
||||
});
|
||||
});
|
||||
@@ -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<void>;
|
||||
}) {
|
||||
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 });
|
||||
}
|
||||
@@ -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 { createNativeTurnAccounting, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent } from "./native-usage-accounting.js";
|
||||
@@ -8091,7 +8092,8 @@ export async function executePaperclipNativeSession(input: {
|
||||
}
|
||||
releaseRetainedNativeSessionCleanup(input.db, {
|
||||
companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
|
||||
});
|
||||
}, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId,
|
||||
issueId: input.execution.binding.issueId });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9473,7 +9475,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
warmNativeSessions.set(warmSessionId, {
|
||||
onReleased: () => releaseRetainedNativeSessionCleanup(input.db, {
|
||||
companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
|
||||
}),
|
||||
}, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId,
|
||||
issueId: input.execution.binding.issueId }),
|
||||
pluginTaskId: usesPluginTask && input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "sandbox"
|
||||
? input.runnerExecutionTarget.sandboxLeaseAcquisition?.providerLeaseId : undefined,
|
||||
runId: input.execution.binding.runId,
|
||||
@@ -11814,7 +11817,8 @@ export async function createRunnerdBackend(input: {
|
||||
initializingSessionToolAuthorities.delete(sessionScopeId);
|
||||
releaseRetainedNativeSessionCleanup(input.db, {
|
||||
companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
|
||||
});
|
||||
}, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId,
|
||||
issueId: input.execution.binding.issueId });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12859,7 +12863,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
})
|
||||
.where(eq(environmentLeases.id, leaseId))
|
||||
.returning({ id: environmentLeases.id })
|
||||
.then((rows) => rows[0] ?? null), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined });
|
||||
.then((rows) => rows[0] ?? null), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined, leaseId });
|
||||
if (!updated) throw new Error("runner_harness_backup_lease_missing");
|
||||
};
|
||||
|
||||
@@ -13278,40 +13282,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);
|
||||
}
|
||||
|
||||
@@ -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<object, Set<(scope: RetainedNativeSessionCleanupScope) => void>>();
|
||||
export function onRetainedNativeSessionCleanupRelease(db: object, listener: (scope: RetainedNativeSessionCleanupScope) => void) {
|
||||
const listeners = new WeakMap<object, Set<(scope: RetainedNativeSessionCleanupScope, source: RetainedNativeSessionCleanupScope) => 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);
|
||||
}
|
||||
Reference in new issue
Block a user