fix(server): wake archive notification delivery from committed changes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/decision-retention-service.test.ts

@@ -1,5 +1,10 @@
+import { sql } from "drizzle-orm";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { registerDecisionArchiveDeliveryWork } from "../services/decision-archive-delivery-work.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
 import { randomUUID } from "node:crypto";
-import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import {
   activityLog,
   agents,
@@ -13,19 +18,23 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro
 import { attentionService } from "../services/attention.js";
 import { decisionRetentionService } from "../services/decision-retention.js";
 
-const support = await getEmbeddedPostgresTestSupport();
+const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RETENTION_TEST_DATABASE_URL");
+const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describePg = support.supported ? describe : describe.skip;
 
 describePg("decision retention", () => {
   let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
   let db: ReturnType<typeof createDb>;
 
+  let stopWorker: (() => Promise<void>) | undefined;
+
   beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-retention-");
-    db = createDb(tempDb.connectionString);
+    if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-retention-");
+    db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
   }, 30_000);
 
   afterEach(async () => {
+    await stopWorker?.(); stopWorker = undefined;
     await db.delete(decisionArchiveNotificationOutbox);
     await db.delete(decisionRetention);
     await db.delete(activityLog);
@@ -34,7 +43,7 @@ describePg("decision retention", () => {
     await db.delete(companies);
   });
 
-  afterAll(async () => tempDb?.cleanup());
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); });
 
   async function seedReviewItems(count: number) {
     const companyId = randomUUID();
@@ -61,7 +70,7 @@ describePg("decision retention", () => {
     await db.insert(issues).values(issueIds.map((id, index) => ({
       id,
       companyId,
-      identifier: `RET-${index + 1}`,
+      identifier: `R${companyId.slice(0, 6)}-${index + 1}`,
       title: `Old review ${index + 1}`,
       status: "in_review",
       priority: "medium",
@@ -90,6 +99,180 @@ describePg("decision retention", () => {
     responsibleUserId: "board-user",
   };
 
+  describe("archive delivery worker", () => {
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let fixture: Awaited<ReturnType<typeof seedReviewItems>>;
+    let service: ReturnType<typeof decisionRetentionService>;
+    let draining: boolean;
+    const errors: unknown[] = [];
+    const notify = vi.fn(async (_batch: unknown) => {});
+
+    beforeEach(async () => {
+      await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+      fixture = await seedReviewItems(1); draining = false; errors.length = 0; notify.mockReset();
+      service = decisionRetentionService(db, { notifyOriginAgent: notify });
+      stopWorker = async () => {
+        await coordinator?.stop(); coordinator = undefined;
+        vi.useRealTimers(); vi.restoreAllMocks();
+        await db.execute(sql`drop trigger if exists archive_delivery_fail on decision_archive_notification_outbox`);
+        await db.execute(sql`drop function if exists archive_delivery_fail()`);
+        expect(errors).toEqual([]);
+      };
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+    });
+    async function start() {
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
+        canReconcile: () => true, onError: error => errors.push(error) });
+      await registerDecisionArchiveDeliveryWork(coordinator, service, () => !draining).ready;
+    }
+    async function archive(source = db) {
+      const feed = await attentionService(source).list(fixture.companyId, { limit: 100 });
+      return decisionRetentionService(source).autoArchive({ companyId: fixture.companyId, items: feed.items });
+    }
+    async function settle(ms = 60_000) {
+      await vi.advanceTimersByTimeAsync(ms);
+      if (!draining) await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 });
+    }
+    async function quiet() {
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(30_000);
+        expect(idleWorkSnapshot().active).toBe(0);
+        expect(coordinator!.nextWakeAt()).toBeNull();
+      }, { timeout: 5000 });
+      const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"), writes = vi.spyOn(db, "update");
+      await settle(60 * 60_000);
+      expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+      reads.mockRestore(); executes.mockRestore(); writes.mockRestore();
+    }
+    async function delivered() {
+      const rows = await db.select().from(decisionArchiveNotificationOutbox);
+      expect(rows.length).toBeGreaterThan(0); expect(rows.every(row => row.status === "delivered")).toBe(true);
+    }
+
+    it("is quiet with configuration, wakes from real auto-archive, and leaves delivered history quiet", async () => {
+      await start(); await quiet();
+      expect(await archive()).toBe(1); await settle(); await delivered(); await quiet();
+      expect(notify).toHaveBeenCalledTimes(1);
+      expect(notify.mock.calls[0][0]).toMatchObject({ companyId: fixture.companyId, agentId: fixture.agentId,
+        items: [expect.objectContaining({ issueId: fixture.issueIds[0], archiveVersion: 1 })] });
+    });
+
+    it("reconstructs pending delivery after restart", async () => {
+      await archive(); await start(); await quiet(); await delivered(); expect(notify).toHaveBeenCalledTimes(1);
+    });
+
+    it("waits until an existing delivery claim expires", async () => {
+      await archive(); const attemptedAt = new Date();
+      await db.update(decisionArchiveNotificationOutbox).set({ status: "delivering", lastAttemptAt: attemptedAt, attemptCount: 1 });
+      await start(); expect(notify).not.toHaveBeenCalled();
+      expect(coordinator!.nextWakeAt()).toBe(attemptedAt.getTime() + 5 * 60_000);
+      const reads = vi.spyOn(db, "select"); await settle(4 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+      await settle(2 * 60_000); await delivered(); await quiet(); expect(notify).toHaveBeenCalledTimes(1);
+    });
+
+    it("recovers a legacy delivering row with no attempt timestamp", async () => {
+      await archive(); await db.update(decisionArchiveNotificationOutbox).set({ status: "delivering" });
+      await start(); await delivered(); await quiet();
+    });
+
+    it("retains first enqueue during drain without passive SQL", async () => {
+      await start(); await quiet(); draining = true;
+      await archive(); await settle();
+      const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update");
+      await settle(60 * 60_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+      reads.mockRestore(); writes.mockRestore(); expect(notify).not.toHaveBeenCalled();
+      draining = false; await settle(); await delivered(); await quiet();
+    });
+
+    it("retries failed delivery through drain and then becomes quiet", async () => {
+      await archive(); notify.mockRejectedValueOnce(new Error("wake unavailable"));
+      await start(); expect(notify).toHaveBeenCalledTimes(1); expect(coordinator!.nextWakeAt()).not.toBeNull();
+      draining = true; await settle(60 * 60_000); expect(notify).toHaveBeenCalledTimes(1);
+      draining = false;
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); await delivered(); }, { timeout: 5000 });
+      await quiet(); expect(notify).toHaveBeenCalledTimes(2);
+    });
+
+    it.each([false, true])("waits for the archive's outer transaction (rollback %s)", async rollback => {
+      await start(); await quiet();
+      let entered!: () => void, release!: () => void;
+      const ready = new Promise<void>(resolve => { entered = resolve; });
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      const aborted = new Error("fixture rollback");
+      const running = db.transaction(async tx => {
+        await archive(tx as unknown as typeof db); entered(); await gate; if (rollback) throw aborted;
+      }).catch(error => { if (error !== aborted) throw error; });
+      await ready;
+      try { await vi.advanceTimersByTimeAsync(60_000); expect(notify).not.toHaveBeenCalled(); }
+      finally { release(); await running; }
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0);
+        if (rollback) expect(await db.select().from(decisionArchiveNotificationOutbox)).toEqual([]);
+        else await delivered();
+      }, { timeout: 5000 });
+      await quiet(); expect(notify).toHaveBeenCalledTimes(rollback ? 0 : 1);
+    });
+
+    it("rolls back archive state when enqueue fails and wakes on a later successful archive", async () => {
+      await start(); await quiet();
+      await db.execute(sql`create function archive_delivery_fail() returns trigger language plpgsql as $$
+        begin raise exception 'archive enqueue fixture failure'; end $$`);
+      await db.execute(sql`create trigger archive_delivery_fail before insert on decision_archive_notification_outbox
+        for each row execute function archive_delivery_fail()`);
+      await expect(archive()).rejects.toThrow();
+      await db.execute(sql`drop trigger archive_delivery_fail on decision_archive_notification_outbox`);
+      await quiet(); expect(notify).not.toHaveBeenCalled();
+      expect(await db.select().from(decisionArchiveNotificationOutbox)).toEqual([]);
+      expect((await db.select().from(decisionRetention))[0].archivedAt).toBeNull();
+      await archive(); await settle(); await delivered(); await quiet();
+    });
+
+    it("keeps an enqueue committed during delivery and drains it on the next pass", async () => {
+      await start(); await quiet();
+      let entered!: () => void, release!: () => void;
+      const ready = new Promise<void>(resolve => { entered = resolve; });
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      notify.mockImplementationOnce(async () => { entered(); await gate; });
+      await archive();
+      await vi.advanceTimersByTimeAsync(30_000); await ready;
+      try {
+        fixture = await seedReviewItems(1); await archive();
+        expect(notify).toHaveBeenCalledTimes(1);
+      } finally { release(); }
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); await delivered(); }, { timeout: 5000 });
+      await quiet(); expect(notify).toHaveBeenCalledTimes(2);
+    });
+
+    it("does not let an older failed batch starve rows beyond the batch limit", async () => {
+      await archive(); const [row] = await db.select().from(decisionArchiveNotificationOutbox);
+      await db.insert(decisionArchiveNotificationOutbox).values(Array.from({ length: 501 }, (_, index) => ({
+        ...row, id: randomUUID(), sourceId: `additional-${index}` })));
+      notify.mockRejectedValueOnce(new Error("first batch unavailable"));
+      await start(); expect(notify.mock.calls[0][0]).toMatchObject({ items: expect.any(Array) });
+      expect((notify.mock.calls[0][0] as { items: unknown[] }).items).toHaveLength(500);
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); await delivered(); }, { timeout: 5000 });
+      await quiet();
+      expect((notify.mock.calls[1][0] as { items: unknown[] }).items).toHaveLength(500);
+      expect((notify.mock.calls[2][0] as { items: unknown[] }).items).toHaveLength(2);
+    });
+
+    it("does not settle a replacement claim from an older delivery result", async () => {
+      await archive();
+      let entered!: () => void, release!: () => void;
+      const ready = new Promise<void>(resolve => { entered = resolve; });
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      notify.mockImplementationOnce(async () => { entered(); await gate; });
+      const first = service.deliverNotifications(); await ready;
+      const replacementAt = new Date(Date.now() + 6 * 60_000);
+      try { await db.update(decisionArchiveNotificationOutbox).set({ lastAttemptAt: replacementAt, attemptCount: 2 }); }
+      finally { release(); }
+      expect((await first).delivered).toBe(0);
+      expect((await db.select().from(decisionArchiveNotificationOutbox))[0]).toMatchObject({ status: "delivering", lastAttemptAt: replacementAt });
+      await start(); expect(coordinator!.nextWakeAt()).toBe(replacementAt.getTime() + 5 * 60_000);
+      await settle(12 * 60_000); await delivered(); await quiet();
+    });
+  });
+
   it("auto-archives, excludes by default, returns archived rows, and revives", async () => {
     const { companyId, issueIds } = await seedReviewItems(1);
     const now = new Date("2026-08-02T00:00:00.000Z");

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -344,6 +344,7 @@ vi.mock("../services/index.js", () => ({
   decisionRetentionService: vi.fn(() => ({
     autoArchive: vi.fn(async () => 0),
     deliverNotifications: vi.fn(async () => ({ notifiedAgents: 0, delivered: 0 })),
+    nextArchiveNotificationAt: vi.fn(async () => null),
   })),
   feedbackService: feedbackServiceFactoryMock,
   bootstrapExecutionPolicyFromEnv: vi.fn(async () => null),

server/src/index.ts

@@ -1,3 +1,4 @@
+import { registerDecisionArchiveDeliveryWork } from "./services/decision-archive-delivery-work.js";
 import { registerToolReviewProjectionWork } from "./services/tool-review-projection-work.js";
 import { registerToolReviewInterruptionWork } from "./services/tool-review-interruption-work.js";
 import { registerToolReviewRecoveryWork } from "./services/tool-review-recovery-work.js";
@@ -1775,8 +1776,7 @@ async function startServerWithDatabaseTeardown(
         });
         archived += await retentionExecutor.autoArchive({ companyId: company.id, items: page.items });
       }
-      const notifications = await retentionExecutor.deliverNotifications();
-      return { archived, ...notifications };
+      return { archived };
     };
     // Startup housekeeping: provider and connection sweeps that every timer tick
     // repeats and that neither serving requests nor run recovery depends on.
@@ -1809,6 +1809,8 @@ async function startServerWithDatabaseTeardown(
       // Retry any orphan sandbox teardown left by a failed acquire before a
       // server restart, so a leaked sandbox does not stay allocated.
       await step("environment_lease_cleanup", startEnvironmentCleanup);
+      await step("archive_notification_worker", () => registerDecisionArchiveDeliveryWork(deliveryWork, retentionExecutor,
+        () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
       await step("retention_sweep", runRetentionSweep);
     };
 

server/src/services/decision-archive-delivery-work.ts

@@ -0,0 +1,17 @@
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { registerDeadlineWork } from "./deadline-work.js";
+import { DECISION_ARCHIVE_RETRY_MS } from "./decision-archive-delivery.js";
+import type { decisionRetentionService } from "./decision-retention.js";
+
+export function registerDecisionArchiveDeliveryWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: Pick<ReturnType<typeof decisionRetentionService>, "deliverNotifications" | "nextArchiveNotificationAt">,
+  canRun: () => boolean,
+) {
+  return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionArchiveDelivery, {
+    retryMs: DECISION_ARCHIVE_RETRY_MS, canRun,
+    run: () => service.deliverNotifications(),
+    next: service.nextArchiveNotificationAt,
+  });
+}

server/src/services/decision-archive-delivery.ts

@@ -0,0 +1,114 @@
+import { and, asc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm";
+import { decisionArchiveNotificationOutbox, type Db } from "@paperclipai/db";
+import type { AttentionSourceKind } from "@paperclipai/shared";
+
+const CLAIM_MS = 5 * 60_000;
+export const DECISION_ARCHIVE_RETRY_MS = 30_000;
+
+export type ArchiveNotificationBatch = {
+  companyId: string;
+  agentId: string;
+  items: Array<{
+    sourceKind: AttentionSourceKind;
+    sourceId: string;
+    archiveVersion: number;
+    issueId: string;
+  }>;
+};
+
+export function decisionArchiveDeliveryService(
+  db: Db,
+  options: { notifyOriginAgent?: (batch: ArchiveNotificationBatch) => Promise<unknown> },
+) {
+  async function deliverNotifications(limit = 500) {
+    if (!options.notifyOriginAgent) return { notifiedAgents: 0, delivered: 0 };
+    const claimAt = new Date();
+    const staleClaimAt = new Date(claimAt.getTime() - CLAIM_MS);
+    await db.update(decisionArchiveNotificationOutbox).set({ status: "pending", updatedAt: claimAt })
+      .where(and(
+        eq(decisionArchiveNotificationOutbox.status, "delivering"),
+        or(isNull(decisionArchiveNotificationOutbox.lastAttemptAt), lte(decisionArchiveNotificationOutbox.lastAttemptAt, staleClaimAt)),
+      ));
+    const rows = await db.transaction(async (tx) => {
+      const pendingIds = await tx.select({ id: decisionArchiveNotificationOutbox.id })
+        .from(decisionArchiveNotificationOutbox)
+        .where(eq(decisionArchiveNotificationOutbox.status, "pending"))
+        .orderBy(sql`${decisionArchiveNotificationOutbox.lastAttemptAt} asc nulls first`,
+          asc(decisionArchiveNotificationOutbox.createdAt), asc(decisionArchiveNotificationOutbox.id))
+        .limit(limit)
+        .then((values) => values.map((value) => value.id));
+      if (pendingIds.length === 0) return [];
+      return tx.update(decisionArchiveNotificationOutbox).set({
+        status: "delivering",
+        lastAttemptAt: claimAt,
+        attemptCount: sql`${decisionArchiveNotificationOutbox.attemptCount} + 1`,
+        updatedAt: claimAt,
+      }).where(and(
+        inArray(decisionArchiveNotificationOutbox.id, pendingIds),
+        eq(decisionArchiveNotificationOutbox.status, "pending"),
+      )).returning();
+    });
+    const groups = new Map<string, typeof rows>();
+    for (const row of rows) {
+      const key = `${row.companyId}:${row.originAgentId}`;
+      const grouped = groups.get(key) ?? [];
+      grouped.push(row);
+      groups.set(key, grouped);
+    }
+    let notifiedAgents = 0;
+    let delivered = 0;
+    for (const grouped of groups.values()) {
+      const first = grouped[0]!;
+      try {
+        await options.notifyOriginAgent({
+          companyId: first.companyId,
+          agentId: first.originAgentId,
+          items: grouped.map((row) => ({
+            sourceKind: row.sourceKind as AttentionSourceKind,
+            sourceId: row.sourceId,
+            archiveVersion: row.archiveVersion,
+            issueId: row.originIssueId,
+          })),
+        });
+        const deliveredAt = new Date();
+        const settled = await db.update(decisionArchiveNotificationOutbox).set({
+          status: "delivered",
+          deliveredAt,
+          lastAttemptAt: deliveredAt,
+          updatedAt: deliveredAt,
+        }).where(and(
+          inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)),
+          eq(decisionArchiveNotificationOutbox.status, "delivering"),
+          eq(decisionArchiveNotificationOutbox.lastAttemptAt, claimAt),
+        )).returning({ id: decisionArchiveNotificationOutbox.id });
+        notifiedAgents += 1;
+        delivered += settled.length;
+      } catch {
+        await db.update(decisionArchiveNotificationOutbox).set({
+          status: "pending",
+          lastAttemptAt: new Date(),
+          updatedAt: new Date(),
+        }).where(and(
+          inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)),
+          eq(decisionArchiveNotificationOutbox.status, "delivering"),
+          eq(decisionArchiveNotificationOutbox.lastAttemptAt, claimAt),
+        ));
+      }
+    }
+    return { notifiedAgents, delivered };
+  }
+
+  async function next() {
+    if (!options.notifyOriginAgent) return null;
+    const [pending] = await db.select({ id: decisionArchiveNotificationOutbox.id })
+      .from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.status, "pending")).limit(1);
+    const [claimed] = await db.select({ attemptedAt: decisionArchiveNotificationOutbox.lastAttemptAt })
+      .from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.status, "delivering"))
+      .orderBy(sql`${decisionArchiveNotificationOutbox.lastAttemptAt} asc nulls first`).limit(1);
+    const pendingAt = pending ? Date.now() + DECISION_ARCHIVE_RETRY_MS : null;
+    const claimAt = claimed ? (claimed.attemptedAt?.getTime() ?? Date.now() - CLAIM_MS) + CLAIM_MS : null;
+    return pendingAt === null ? claimAt : claimAt === null ? pendingAt : Math.min(pendingAt, claimAt);
+  }
+
+  return { deliver: deliverNotifications, next };
+}

server/src/services/decision-retention.ts

@@ -1,5 +1,7 @@
+import { decisionArchiveDeliveryService, type ArchiveNotificationBatch } from "./decision-archive-delivery.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { createHash } from "node:crypto";
-import { and, asc, eq, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm";
+import { and, eq, inArray, isNotNull, isNull, lte, ne, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
   agents,
@@ -31,16 +33,7 @@ export const DEFAULT_DECISION_ARCHIVE_DAYS = 90;
 const DAY_MS = 86_400_000;
 
 export type DecisionRetentionState = typeof decisionRetention.$inferSelect;
-export type ArchiveNotificationBatch = {
-  companyId: string;
-  agentId: string;
-  items: Array<{
-    sourceKind: AttentionSourceKind;
-    sourceId: string;
-    archiveVersion: number;
-    issueId: string;
-  }>;
-};
+export type { ArchiveNotificationBatch } from "./decision-archive-delivery.js";
 
 function sourceKey(sourceKind: string, sourceId: string) {
   return `${sourceKind}:${sourceId}`;
@@ -371,6 +364,7 @@ export function decisionRetentionService(
         archived += 1;
         const origin = await resolveOrigin(txDb, input.companyId, item.sourceKind, item.subject.id);
         if (origin) {
+          await notifyDeliveryWork(txDb, DELIVERY_QUEUES.decisionArchiveDelivery);
           await txDb.insert(decisionArchiveNotificationOutbox).values({
             companyId: input.companyId,
             sourceKind: item.sourceKind,
@@ -394,80 +388,7 @@ export function decisionRetentionService(
     return archived;
   }
 
-  async function deliverNotifications(limit = 500) {
-    if (!options.notifyOriginAgent) return { notifiedAgents: 0, delivered: 0 };
-    const claimAt = new Date();
-    const staleClaimAt = new Date(claimAt.getTime() - 5 * 60_000);
-    await db.update(decisionArchiveNotificationOutbox).set({ status: "pending", updatedAt: claimAt })
-      .where(and(
-        eq(decisionArchiveNotificationOutbox.status, "delivering"),
-        or(isNull(decisionArchiveNotificationOutbox.lastAttemptAt), lte(decisionArchiveNotificationOutbox.lastAttemptAt, staleClaimAt)),
-      ));
-    const rows = await db.transaction(async (tx) => {
-      const pendingIds = await tx.select({ id: decisionArchiveNotificationOutbox.id })
-        .from(decisionArchiveNotificationOutbox)
-        .where(eq(decisionArchiveNotificationOutbox.status, "pending"))
-        .orderBy(asc(decisionArchiveNotificationOutbox.createdAt), asc(decisionArchiveNotificationOutbox.id))
-        .limit(limit)
-        .then((values) => values.map((value) => value.id));
-      if (pendingIds.length === 0) return [];
-      return tx.update(decisionArchiveNotificationOutbox).set({
-        status: "delivering",
-        lastAttemptAt: claimAt,
-        attemptCount: sql`${decisionArchiveNotificationOutbox.attemptCount} + 1`,
-        updatedAt: claimAt,
-      }).where(and(
-        inArray(decisionArchiveNotificationOutbox.id, pendingIds),
-        eq(decisionArchiveNotificationOutbox.status, "pending"),
-      )).returning();
-    });
-    const groups = new Map<string, typeof rows>();
-    for (const row of rows) {
-      const key = `${row.companyId}:${row.originAgentId}`;
-      const grouped = groups.get(key) ?? [];
-      grouped.push(row);
-      groups.set(key, grouped);
-    }
-    let notifiedAgents = 0;
-    let delivered = 0;
-    for (const grouped of groups.values()) {
-      const first = grouped[0]!;
-      try {
-        await options.notifyOriginAgent({
-          companyId: first.companyId,
-          agentId: first.originAgentId,
-          items: grouped.map((row) => ({
-            sourceKind: row.sourceKind as AttentionSourceKind,
-            sourceId: row.sourceId,
-            archiveVersion: row.archiveVersion,
-            issueId: row.originIssueId,
-          })),
-        });
-        const deliveredAt = new Date();
-        await db.update(decisionArchiveNotificationOutbox).set({
-          status: "delivered",
-          deliveredAt,
-          lastAttemptAt: deliveredAt,
-          updatedAt: deliveredAt,
-        }).where(and(
-          inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)),
-          eq(decisionArchiveNotificationOutbox.status, "delivering"),
-        ));
-        notifiedAgents += 1;
-        delivered += grouped.length;
-      } catch {
-        await db.update(decisionArchiveNotificationOutbox).set({
-          status: "pending",
-          lastAttemptAt: new Date(),
-          updatedAt: new Date(),
-        }).where(and(
-          inArray(decisionArchiveNotificationOutbox.id, grouped.map((row) => row.id)),
-          eq(decisionArchiveNotificationOutbox.status, "delivering"),
-        ));
-      }
-    }
-    return { notifiedAgents, delivered };
-  }
+  const archiveDelivery = decisionArchiveDeliveryService(db, options);
 
   return {
     syncItems,
@@ -477,6 +398,7 @@ export function decisionRetentionService(
     revive: (input: Omit<Parameters<typeof changeArchived>[0], "archived">) => changeArchived({ ...input, archived: false }),
     archiveReviewedManifest,
     autoArchive,
-    deliverNotifications,
+    deliverNotifications: archiveDelivery.deliver,
+    nextArchiveNotificationAt: archiveDelivery.next,
   };
 }

server/src/services/delivery-work-notifications.ts

@@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = {
   executionStatus: "execution-status",
   executionDisposition: "execution-disposition",
   environmentCleanup: "environment-cleanup",
+  decisionArchiveDelivery: "decision-archive-delivery",
   toolReviewProjection: "tool-review-projection",
   toolReviewInterruption: "tool-review-interruption",
   toolReviewRecovery: "tool-review-recovery",