fix(server): schedule approval retention from durable source deadlines

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-24 14:43:07 -07:00
1 parent f240381fd8
commit ea3dd40cdd
27 files changed
+893 -150

No files matched your search

@@ -0,0 +1,325 @@
import { randomUUID } from "node:crypto";
import { eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { activityLog, agents, approvals, chatConversations, chatEndpoints, companies, createDb, decisionArchiveNotificationOutbox, decisionRetention, issueApprovals, issues, toolApplications, toolConnections, type Db } from "@paperclipai/db";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { approvalRecords } from "../services/approvals.js";
import { approvalRetentionService, registerApprovalRetentionWork } from "../services/approval-retention.js";
import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js";
import { attentionService } from "../services/attention.js";
import { issueApprovalService } from "../services/issue-approvals.js";
import { companyService } from "../services/companies.js";
import { issueService } from "../services/issues.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_APPROVAL_RETENTION_TEST_DATABASE_URL");
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
(support.supported ? describe : describe.skip)("approval retention worker", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>, companyId: string;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let draining: boolean;
const errors: unknown[] = [];
beforeAll(async () => {
if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("approval-retention-");
db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
}, 30_000);
beforeEach(async () => {
companyId = randomUUID(); draining = false; errors.length = 0;
await db.insert(companies).values({ id: companyId, name: "Approval retention", issuePrefix: `A${companyId.slice(0, 6)}` });
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
});
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined;
vi.useRealTimers(); vi.restoreAllMocks();
await db.execute(sql`drop trigger if exists approval_retention_fail on decision_retention`);
await db.execute(sql`drop function if exists approval_retention_fail()`);
await db.delete(decisionArchiveNotificationOutbox); await db.delete(decisionRetention);
await db.delete(activityLog); await db.delete(issueApprovals); await db.delete(approvals);
await db.delete(chatConversations); await db.delete(chatEndpoints);
await db.delete(toolConnections); await db.delete(toolApplications);
await db.delete(issues); await db.delete(agents); await db.delete(companies);
expect(errors).toEqual([]);
});
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
const create = (at = new Date(), database: Db = db) => approvalRecords(database).create(companyId, {
type: "request_board_approval", payload: {}, createdAt: at, updatedAt: at,
});
const old = () => new Date(Date.now() - delay - 60_000);
const state = (id: string) => decisionRetentionService(db).getState(companyId, "approval", id);
const authActor = () => ({ type: "board" as const, source: "local_implicit" as const, userId: "board-user", companyIds: [companyId], isInstanceAdmin: true });
const actor = { actorType: "user" as const, actorId: "board-user", agentId: null, userId: "board-user", runId: null, agentApiKeyId: null, responsibleUserId: "board-user" };
const mutation = (id: string) => ({ companyId, sourceKind: "approval" as const, sourceId: id, authActor: authActor(), actor });
async function start(service = approvalRetentionService(db)) {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) });
await registerApprovalRetentionWork(coordinator, service, () => !draining).ready;
return service;
}
async function settle() {
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(30_000);
expect(idleWorkSnapshot().active).toBe(0);
}, { timeout: 5000 });
}
async function quiet() {
await settle();
expect(coordinator!.nextWakeAt()).toBeNull();
const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"), writes = vi.spyOn(db, "transaction");
await vi.advanceTimersByTimeAsync(300_000);
expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
reads.mockRestore(); executes.mockRestore(); writes.mockRestore();
}
async function archived(id: string) {
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(30_000);
expect((await state(id))?.archivedAt).toBeInstanceOf(Date);
}, { timeout: 5000 });
}
async function issue(hiddenAt: Date | null = null) {
const [row] = await db.insert(issues).values({ companyId, title: "Approval target", status: "todo", hiddenAt }).returning();
return row;
}
it("is quiet with no approvals and archives unseen durable approvals at their actual deadline", async () => {
await start(); await quiet();
const deadline = Date.now() + 120_000;
const row = await create(new Date(deadline - delay));
await settle();
expect(await state(row.id)).toBeNull();
expect(coordinator!.nextWakeAt()).toBe(deadline);
const reads = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(30_000);
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
await vi.advanceTimersByTimeAsync(120_000);
await archived(row.id); await quiet();
expect((await db.select().from(approvals).where(eq(approvals.id, row.id)))[0].status).toBe("pending");
});
it("recovers old unseen approvals on startup and leaves completed history quiet", async () => {
const due = await create(old()), completed = await create(old());
await approvalRecords(db).resolveApproval(completed.id, "approved", "board-user", null);
await start(); await archived(due.id); await quiet();
expect(await state(completed.id)).toBeNull();
});
it("disarms on revision and moves the deadline when resubmitted", async () => {
await start(); const row = await create(); await settle();
await approvalRecords(db).requestRevision(row.id, "board-user"); await quiet();
const resubmitted = await approvalRecords(db).resubmit(row.id); await settle();
expect(coordinator!.nextWakeAt()).toBe(resubmitted.updatedAt.getTime() + delay);
await approvalRecords(db).cancel(row.id); await quiet();
});
it("honors keep and wakes when an old approval is unkept", async () => {
const row = await create(old());
await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: true });
await start(); await quiet(); expect((await state(row.id))?.archivedAt).toBeNull();
await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: false });
await archived(row.id); await quiet();
});
it("disarms on manual archive and reactivates on revive", async () => {
const row = await create(old());
await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
await decisionRetentionService(db).archive(mutation(row.id));
await start(); await quiet();
await decisionRetentionService(db).revive(mutation(row.id));
await archived(row.id); await quiet();
expect((await state(row.id))?.archivedReason).toBe("idle_ttl");
});
it("waits for inactive companies and wakes on company activation", async () => {
const row = await create(old());
await companyService(db).update(companyId, { status: "paused" });
await start(); await quiet(); expect(await state(row.id)).toBeNull();
await companyService(db).update(companyId, { status: "active" });
await archived(row.id); await quiet();
});
it("excludes hidden linked targets and wakes after unlink", async () => {
const row = await create(old()), target = await issue(new Date());
await issueApprovalService(db).link(target.id, row.id);
await start(); await quiet(); expect(await state(row.id)).toBeNull();
await issueApprovalService(db).unlink(target.id, row.id);
await archived(row.id); await quiet();
});
it("wakes when the linked target becomes visible", async () => {
const row = await create(old()), target = await issue(new Date());
await issueApprovalService(db).link(target.id, row.id);
await start(); await quiet();
await issueService(db).update(target.id, { hiddenAt: null });
await archived(row.id); await quiet();
});
it("wakes after deletion cascades the excluded target link", async () => {
const row = await create(old()), target = await issue(new Date());
await issueApprovalService(db).link(target.id, row.id);
await start(); await quiet();
await issueService(db).remove(target.id);
await archived(row.id); await quiet();
});
it("preserves normal attention reads while excluding approval retention maintenance", async () => {
const row = await create(old());
const excluded = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, excludeSources: ["approval"] });
expect(excluded.items.some(item => item.sourceKind === "approval")).toBe(false);
expect(await state(row.id)).toBeNull();
const visible = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
expect(visible.items.some(item => item.subject.id === row.id)).toBe(true);
expect(await state(row.id)).not.toBeNull();
});
it("does not publish work before an outer transaction commits", async () => {
await start(); await quiet();
let id = "";
await db.transaction(async tx => {
id = (await create(old(), tx as unknown as Db)).id;
await vi.advanceTimersByTimeAsync(60_000);
expect(await state(id)).toBeNull();
});
await archived(id); await quiet();
});
it("discharges rolled back nested producers without archiving", async () => {
await start(); await quiet();
await expect(db.transaction(async tx => {
await create(old(), tx as unknown as Db); throw new Error("rollback");
})).rejects.toThrow("rollback");
await quiet(); expect(await db.select().from(approvals)).toEqual([]);
});
it("retains the first producer during drain and processes it after resume", async () => {
await start(); await quiet(); draining = true;
const row = await create(old());
await vi.advanceTimersByTimeAsync(90_000); expect(await state(row.id)).toBeNull();
const reads = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(90_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
draining = false; await archived(row.id); await quiet();
});
it("rechecks current source activity after waiting on an approval writer", async () => {
const row = await create(old());
let unlock!: () => void, locked!: () => void;
const barrier = new Promise<void>(resolve => { unlock = resolve; });
const admitted = new Promise<void>(resolve => { locked = resolve; });
const write = db.transaction(async tx => {
await tx.execute(sql`select id from approvals where id = ${row.id} for update`);
locked(); await barrier;
await approvalRecords(tx as unknown as Db).requestRevision(row.id, "board-user");
await approvalRecords(tx as unknown as Db).resubmit(row.id);
});
await admitted;
const sweep = approvalRetentionService(db).sweep();
await vi.advanceTimersByTimeAsync(100);
unlock(); await write; await sweep;
expect(await state(row.id)).toBeNull();
});
async function failArchives() {
await db.execute(sql`create function approval_retention_fail() returns trigger language plpgsql as $$
begin
if new.archived_at is not null and exists (
select 1 from approvals where id::text = new.source_id and payload->>'retentionTestFailure' = 'true'
) then raise exception 'injected approval archive failure'; end if;
return new;
end $$`);
await db.execute(sql`create trigger approval_retention_fail before update on decision_retention
for each row execute function approval_retention_fail()`);
}
it("retains failed durable work across drain and resumes recovery", async () => {
const row = await approvalRecords(db).create(companyId, { type: "request_board_approval",
payload: { retentionTestFailure: true }, updatedAt: old() });
await failArchives();
await start();
expect(errors.length).toBeGreaterThan(0); errors.length = 0;
draining = true;
await vi.advanceTimersByTimeAsync(60_000);
const reads = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
await db.execute(sql`drop trigger approval_retention_fail on decision_retention`);
draining = false; await archived(row.id); await quiet();
});
it("advances bounded pages past a failed row without losing its retry", async () => {
const at = old();
await db.insert(approvals).values(Array.from({ length: 101 }, (_, i) => ({ companyId,
type: "request_board_approval", payload: { retentionTestFailure: i === 0 },
createdAt: at, updatedAt: new Date(at.getTime() + i),
})));
await failArchives();
const service = approvalRetentionService(db);
await expect(service.sweep()).rejects.toThrow("Failed query");
expect(await service.sweep()).toEqual({ archived: 1 });
expect((await db.select().from(decisionRetention)).filter(row => row.archivedAt)).toHaveLength(100);
expect(await service.next()).not.toBeNull();
await db.execute(sql`drop trigger approval_retention_fail on decision_retention`);
expect(await service.sweep()).toEqual({ archived: 1 }); expect(await service.next()).toBeNull();
});
it("reconstructs startup obligations while paused and performs no passive SQL", async () => {
const row = await create(old()); draining = true;
await start();
const reads = vi.spyOn(db, "select");
await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
expect(await state(row.id)).toBeNull();
draining = false; await archived(row.id); await quiet();
});
it("matches the feed's first linked issue rather than any visible link", async () => {
const row = await create(old()), one = await issue(), two = await issue();
const [first, second] = [one, two].sort((a, b) => a.id.localeCompare(b.id));
await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, first.id));
await issueApprovalService(db).linkManyForApproval(row.id, [second.id, first.id]);
expect((await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true })).items
.some(item => item.subject.id === row.id)).toBe(false);
await start(); await quiet();
await issueApprovalService(db).unlink(first.id, row.id);
await archived(row.id); await quiet();
});
it("joins an admitted sweep on shutdown and skips subsequent deadline queries", async () => {
const row = await create(old());
const service = approvalRetentionService(db), original = service.sweep;
let release!: () => void, admitted!: () => void;
const barrier = new Promise<void>(resolve => { release = resolve; });
const admission = new Promise<void>(resolve => { admitted = resolve; });
vi.spyOn(service, "sweep").mockImplementation(async (now, signal) => {
admitted(); await barrier; return original(now, signal);
});
const next = vi.spyOn(service, "next");
const starting = start(service); await admission;
let stopped = false;
const stopping = coordinator!.stop().then(() => { stopped = true; });
await vi.advanceTimersByTimeAsync(100); expect(stopped).toBe(false);
release(); await starting; await stopping; coordinator = undefined;
expect(next).not.toHaveBeenCalled(); expect(await state(row.id)).toBeNull();
await start(); await archived(row.id); await quiet();
});
it("uses the shared feed eligibility for a waiting Slack conversation with a pending approval", async () => {
const [agent] = await db.insert(agents).values({ companyId, name: "Origin", role: "engineer", status: "active", adapterType: "codex_local" }).returning();
const [app] = await db.insert(toolApplications).values({ companyId, name: "Slack", type: "chat" }).returning();
const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app.id, name: "Slack", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning();
const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId: connection.id, provider: "slack", publicId: randomUUID(), assignedAgentId: agent.id, status: "active" }).returning();
const target = await issue();
await db.update(issues).set({ status: "in_review", assigneeAgentId: agent.id }).where(eq(issues.id, target.id));
await db.insert(chatConversations).values({ companyId, endpointId: endpoint.id, issueId: target.id, externalConversationId: randomUUID(), externalLabel: "Waiting", state: "waiting" });
const row = await approvalRecords(db).create(companyId, { type: "request_board_approval", payload: {}, requestedByAgentId: agent.id, updatedAt: old() });
await issueApprovalService(db).link(target.id, row.id);
const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
expect(feed.items.some(item => item.subject.id === row.id)).toBe(true);
await start(); await archived(row.id); await quiet();
expect(await db.select().from(decisionArchiveNotificationOutbox)).toMatchObject([
{ sourceId: row.id, originIssueId: target.id, originAgentId: agent.id, status: "pending" },
]);
});
});
@@ -60,14 +60,12 @@ function createDbStub(selectResults: ApprovalRecord[][], updateResults: Approval
const set = vi.fn(() => ({ where: updateWhere }));
const update = vi.fn(() => ({ set }));
return {
db: { select, update, transaction: vi.fn(async (callback: (db: unknown) => Promise<unknown>) => {
inTransaction = true;
try { return await callback({ select, update }); } finally { inTransaction = false; }
}) },
selectWhere,
returning,
};
const db = { select, update, transaction: vi.fn(async (callback: (db: unknown) => Promise<unknown>) => {
const outerTransaction = inTransaction;
inTransaction = true;
try { return await callback(db); } finally { inTransaction = outerTransaction; }
}) };
return { db, selectWhere, returning };
}
describe("approvalService resolution idempotency", () => {
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { createAgentLifecycle } from "../services/agent-lifecycle.js";
import { randomUUID } from "node:crypto";
import express from "express";
@@ -47,7 +48,8 @@ import { listAttentionExhaustedRuns } from "../services/attention-exhausted-runs
import { agentService } from "../services/agents.js";
import { ROUTABLE_BLOCKED_ROLLOUT_AT } from "../services/routable-blocked.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ATTENTION_TEST_DATABASE_URL");
const embeddedPostgresSupport = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
@@ -61,8 +63,8 @@ describeEmbeddedPostgres("attention service", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-attention-service-");
db = createDb(tempDb.connectionString);
if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-attention-service-");
db = createDb(externalUrl ?? tempDb!.connectionString);
}, 30_000);
afterEach(async () => {
@@ -98,6 +100,7 @@ describeEmbeddedPostgres("attention service", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
+7 -3
View File
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { randomUUID } from "node:crypto";
import { eq } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -26,7 +27,8 @@ vi.mock("../services/activity-log.js", () => ({
// Budget behavior is exercised against PostgreSQL below and in
// cost-accounting-reliability.test.ts, including transaction and retry paths.
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_BUDGETS_TEST_DATABASE_URL");
const embeddedPostgresSupport = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
describeEmbeddedPostgres("budgetService release gate enforcement", () => {
@@ -34,8 +36,9 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-budgets-service-");
db = createDb(tempDb.connectionString);
await import("../services/agent-lifecycle.js");
if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-budgets-service-");
db = createDb(externalUrl ?? tempDb!.connectionString);
}, 20_000);
afterEach(async () => {
@@ -50,6 +53,7 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
@@ -1,3 +1,4 @@
import { approvalRetentionService } from "../services/approval-retention.js";
import { logger } from "../middleware/logger.js";
import { getStartupRecoveryState } from "../startup-recovery-state.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
@@ -321,6 +322,11 @@ vi.mock("../realtime/live-events-ws.js", () => ({
setupLiveEventsWebSocketServer: vi.fn(),
}));
vi.mock("../services/approval-retention.js", async importOriginal => ({
...await importOriginal<typeof import("../services/approval-retention.js")>(),
approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })),
}));
vi.mock("../services/index.js", () => ({
backfillLegacyToolOAuthTokens: vi.fn(async () => ({
scannedConnections: 0,
@@ -519,6 +525,18 @@ describe("startServer feedback export wiring", () => {
process.env.BETTER_AUTH_SECRET = "test-secret";
});
it.each([false, true])("registers approval retention only when maintenance is enabled (%s)", async enabled => {
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
await startServer(); await settleStartupHousekeeping();
const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.approvalRetention);
expect(Boolean(registration)).toBe(enabled);
if (enabled) {
const service = vi.mocked(approvalRetentionService).mock.results.at(-1)!.value;
await registration![1].run(new AbortController().signal);
expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled();
}
});
it("registers durable case handoffs independently of the heartbeat scheduler", async () => {
await startServer();
expect(caseHandoffDeliveryServiceFactoryMock).toHaveBeenCalledWith(createDbMock.mock.results[0]!.value, { heartbeat: heartbeatServiceMock });